/proc/self/cwd/source/common/router/scoped_rds.cc
Line | Count | Source (jump to first uncovered line) |
1 | | #include "source/common/router/scoped_rds.h" |
2 | | |
3 | | #include <memory> |
4 | | |
5 | | #include "envoy/admin/v3/config_dump.pb.h" |
6 | | #include "envoy/config/core/v3/config_source.pb.h" |
7 | | #include "envoy/config/route/v3/scoped_route.pb.h" |
8 | | #include "envoy/extensions/filters/network/http_connection_manager/v3/http_connection_manager.pb.h" |
9 | | #include "envoy/service/discovery/v3/discovery.pb.h" |
10 | | |
11 | | #include "source/common/common/assert.h" |
12 | | #include "source/common/common/cleanup.h" |
13 | | #include "source/common/common/logger.h" |
14 | | #include "source/common/common/utility.h" |
15 | | #include "source/common/config/api_version.h" |
16 | | #include "source/common/config/resource_name.h" |
17 | | #include "source/common/config/xds_resource.h" |
18 | | #include "source/common/init/manager_impl.h" |
19 | | #include "source/common/init/watcher_impl.h" |
20 | | #include "source/common/protobuf/utility.h" |
21 | | #include "source/common/router/rds_impl.h" |
22 | | #include "source/common/router/scoped_config_impl.h" |
23 | | |
24 | | #include "absl/strings/str_join.h" |
25 | | |
26 | | // Types are deeply nested under Envoy::Config::ConfigProvider; use 'using-directives' across all |
27 | | // ConfigProvider related types for consistency. |
28 | | using Envoy::Config::ConfigProvider; |
29 | | using Envoy::Config::ConfigProviderInstanceType; |
30 | | using Envoy::Config::ConfigProviderManager; |
31 | | using Envoy::Config::ConfigProviderPtr; |
32 | | using Envoy::Config::ScopedResume; |
33 | | |
34 | | namespace Envoy { |
35 | | namespace Router { |
36 | | namespace ScopedRoutesConfigProviderUtil { |
37 | | ConfigProviderPtr create( |
38 | | const envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager& |
39 | | config, |
40 | | Server::Configuration::ServerFactoryContext& factory_context, Init::Manager& init_manager, |
41 | 101 | const std::string& stat_prefix, ConfigProviderManager& scoped_routes_config_provider_manager) { |
42 | 101 | ASSERT(config.route_specifier_case() == |
43 | 101 | envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager:: |
44 | 101 | RouteSpecifierCase::kScopedRoutes); |
45 | 101 | OptionalHttpFilters optional_http_filters; |
46 | 101 | auto& filters = config.http_filters(); |
47 | 112 | for (const auto& filter : filters) { |
48 | 112 | if (filter.is_optional()) { |
49 | 101 | optional_http_filters.insert(filter.name()); |
50 | 101 | } |
51 | 112 | } |
52 | 101 | switch (config.scoped_routes().config_specifier_case()) { |
53 | 92 | case envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes:: |
54 | 92 | ConfigSpecifierCase::kScopedRouteConfigurationsList: { |
55 | 92 | const envoy::extensions::filters::network::http_connection_manager::v3:: |
56 | 92 | ScopedRouteConfigurationsList& scoped_route_list = |
57 | 92 | config.scoped_routes().scoped_route_configurations_list(); |
58 | 92 | return scoped_routes_config_provider_manager.createStaticConfigProvider( |
59 | 92 | RepeatedPtrUtil::convertToConstMessagePtrContainer< |
60 | 92 | envoy::config::route::v3::ScopedRouteConfiguration, |
61 | 92 | ProtobufTypes::ConstMessagePtrVector>(scoped_route_list.scoped_route_configurations()), |
62 | 92 | factory_context, |
63 | 92 | ScopedRoutesConfigProviderManagerOptArg(config.scoped_routes().name(), |
64 | 92 | config.scoped_routes().rds_config_source(), |
65 | 92 | optional_http_filters)); |
66 | 0 | } |
67 | 9 | case envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes:: |
68 | 9 | ConfigSpecifierCase::kScopedRds: |
69 | 9 | return scoped_routes_config_provider_manager.createXdsConfigProvider( |
70 | 9 | config.scoped_routes().scoped_rds(), factory_context, init_manager, stat_prefix, |
71 | 9 | ScopedRoutesConfigProviderManagerOptArg(config.scoped_routes().name(), |
72 | 9 | config.scoped_routes().rds_config_source(), |
73 | 9 | optional_http_filters)); |
74 | 0 | case envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes:: |
75 | 0 | ConfigSpecifierCase::CONFIG_SPECIFIER_NOT_SET: |
76 | 0 | PANIC("not implemented"); |
77 | 101 | } |
78 | 0 | PANIC_DUE_TO_CORRUPT_ENUM; |
79 | 0 | } |
80 | | |
81 | | ScopeKeyBuilderPtr createScopeKeyBuilder( |
82 | | const envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager& |
83 | 81 | config) { |
84 | 81 | ASSERT(config.route_specifier_case() == |
85 | 81 | envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager:: |
86 | 81 | RouteSpecifierCase::kScopedRoutes); |
87 | 81 | auto scope_key_builder = config.scoped_routes().scope_key_builder(); |
88 | 81 | return std::make_unique<ScopeKeyBuilderImpl>(std::move(scope_key_builder)); |
89 | 81 | } |
90 | | |
91 | | } // namespace ScopedRoutesConfigProviderUtil |
92 | | |
93 | | namespace { |
94 | | |
95 | | std::vector<ScopedRouteInfoConstSharedPtr> |
96 | | makeScopedRouteInfos(ProtobufTypes::ConstMessagePtrVector&& config_protos, |
97 | | Server::Configuration::ServerFactoryContext& factory_context, |
98 | | ScopedRoutesConfigProviderManager& config_provider_manager, |
99 | 92 | const OptionalHttpFilters& optional_http_filters) { |
100 | 92 | std::vector<ScopedRouteInfoConstSharedPtr> scopes; |
101 | 1.01k | for (std::unique_ptr<const Protobuf::Message>& config_proto : config_protos) { |
102 | 1.01k | auto scoped_route_config = |
103 | 1.01k | MessageUtil::downcastAndValidate<const envoy::config::route::v3::ScopedRouteConfiguration&>( |
104 | 1.01k | *config_proto, factory_context.messageValidationContext().staticValidationVisitor()); |
105 | 1.01k | if (!scoped_route_config.route_configuration_name().empty()) { |
106 | 6 | throwEnvoyExceptionOrPanic( |
107 | 6 | "Fetching routes via RDS (route_configuration_name) is not supported " |
108 | 6 | "with inline scoped routes."); |
109 | 6 | } |
110 | 1.00k | if (!scoped_route_config.has_route_configuration()) { |
111 | 0 | throwEnvoyExceptionOrPanic( |
112 | 0 | "You must specify a route_configuration with inline scoped routes."); |
113 | 0 | } |
114 | 1.00k | RouteConfigProviderPtr route_config_provider = |
115 | 1.00k | config_provider_manager.routeConfigProviderManager().createStaticRouteConfigProvider( |
116 | 1.00k | scoped_route_config.route_configuration(), optional_http_filters, factory_context, |
117 | 1.00k | factory_context.messageValidationContext().staticValidationVisitor()); |
118 | 1.00k | scopes.push_back(std::make_shared<const ScopedRouteInfo>(scoped_route_config, |
119 | 1.00k | route_config_provider->configCast())); |
120 | 1.00k | } |
121 | | |
122 | 86 | return scopes; |
123 | 92 | } |
124 | | |
125 | | } // namespace |
126 | | |
127 | | InlineScopedRoutesConfigProvider::InlineScopedRoutesConfigProvider( |
128 | | ProtobufTypes::ConstMessagePtrVector&& config_protos, std::string name, |
129 | | Server::Configuration::ServerFactoryContext& factory_context, |
130 | | ScopedRoutesConfigProviderManager& config_provider_manager, |
131 | | envoy::config::core::v3::ConfigSource rds_config_source, |
132 | | const OptionalHttpFilters& optional_http_filters) |
133 | | : Envoy::Config::ImmutableConfigProviderBase(factory_context, config_provider_manager, |
134 | | ConfigProviderInstanceType::Inline, |
135 | | ConfigProvider::ApiType::Delta), |
136 | | name_(std::move(name)), |
137 | | scopes_(makeScopedRouteInfos(std::move(config_protos), factory_context, |
138 | | config_provider_manager, optional_http_filters)), |
139 | | config_(std::make_shared<ScopedConfigImpl>(scopes_)), |
140 | 92 | rds_config_source_(std::move(rds_config_source)) {} |
141 | | |
142 | | ScopedRdsConfigSubscription::ScopedRdsConfigSubscription( |
143 | | const envoy::extensions::filters::network::http_connection_manager::v3::ScopedRds& scoped_rds, |
144 | | const OptionalHttpFilters& optional_http_filters, const uint64_t manager_identifier, |
145 | | const std::string& name, Server::Configuration::ServerFactoryContext& factory_context, |
146 | | const std::string& stat_prefix, envoy::config::core::v3::ConfigSource rds_config_source, |
147 | | RouteConfigProviderManager& route_config_provider_manager, |
148 | | ScopedRoutesConfigProviderManager& config_provider_manager) |
149 | | : DeltaConfigSubscriptionInstance("SRDS", manager_identifier, config_provider_manager, |
150 | | factory_context), |
151 | | Envoy::Config::SubscriptionBase<envoy::config::route::v3::ScopedRouteConfiguration>( |
152 | | factory_context.messageValidationContext().dynamicValidationVisitor(), "name"), |
153 | | factory_context_(factory_context), name_(name), |
154 | | scope_(factory_context.scope().createScope(stat_prefix + "scoped_rds." + name + ".")), |
155 | | stats_({ALL_SCOPED_RDS_STATS(POOL_COUNTER(*scope_), POOL_GAUGE(*scope_))}), |
156 | | rds_config_source_(std::move(rds_config_source)), stat_prefix_(stat_prefix), |
157 | | route_config_provider_manager_(route_config_provider_manager), |
158 | 7 | optional_http_filters_(optional_http_filters) { |
159 | 7 | const auto resource_name = getResourceName(); |
160 | 7 | if (scoped_rds.srds_resources_locator().empty()) { |
161 | 6 | subscription_ = |
162 | 6 | factory_context.clusterManager().subscriptionFactory().subscriptionFromConfigSource( |
163 | 6 | scoped_rds.scoped_rds_config_source(), Grpc::Common::typeUrl(resource_name), *scope_, |
164 | 6 | *this, resource_decoder_, {}); |
165 | 6 | } else { |
166 | 1 | const auto srds_resources_locator = |
167 | 1 | Envoy::Config::XdsResourceIdentifier::decodeUrl(scoped_rds.srds_resources_locator()); |
168 | 1 | subscription_ = |
169 | 1 | factory_context.clusterManager().subscriptionFactory().collectionSubscriptionFromUrl( |
170 | 1 | srds_resources_locator, scoped_rds.scoped_rds_config_source(), resource_name, *scope_, |
171 | 1 | *this, resource_decoder_); |
172 | 1 | } |
173 | | |
174 | | // TODO(tony612): consider not using the callback here. |
175 | 7 | initialize([]() -> Envoy::Config::ConfigProvider::ConfigConstSharedPtr { |
176 | 6 | return std::make_shared<ScopedConfigImpl>(); |
177 | 6 | }); |
178 | 7 | } |
179 | | |
180 | | // Constructor for RdsRouteConfigProviderHelper when scope is eager loading. |
181 | | // Initialize RdsRouteConfigProvider by default. |
182 | | ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper::RdsRouteConfigProviderHelper( |
183 | | ScopedRdsConfigSubscription& parent, std::string scope_name, |
184 | | envoy::extensions::filters::network::http_connection_manager::v3::Rds& rds, |
185 | | Init::Manager& init_manager) |
186 | 0 | : parent_(parent), scope_name_(scope_name), on_demand_(false) { |
187 | 0 | initRdsConfigProvider(rds, init_manager); |
188 | 0 | } |
189 | | |
190 | | // Constructor for RdsRouteConfigProviderHelper when scope is on demand. |
191 | | // Leave the RdsRouteConfigProvider uninitialized. |
192 | | ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper::RdsRouteConfigProviderHelper( |
193 | | ScopedRdsConfigSubscription& parent, std::string scope_name) |
194 | 0 | : parent_(parent), scope_name_(scope_name), on_demand_(true) { |
195 | 0 | parent_.stats_.on_demand_scopes_.inc(); |
196 | 0 | } |
197 | | |
198 | | // When on demand callback is received from main thread, there are 4 cases. |
199 | | // 1. Scope is not found, post a scope not found callback back to worker thread. |
200 | | // 2. Scope is found but route provider has not been initialized, create route provider. |
201 | | // 3. After route provider has been initialized, if RouteConfiguration has been fetched, |
202 | | // post scope found callback to worker thread. |
203 | | // 4. After route provider has been initialized, if RouteConfiguration is null, |
204 | | // cache the callback and wait for RouteConfiguration to come. |
205 | | void ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper::addOnDemandUpdateCallback( |
206 | 0 | std::function<void()> callback) { |
207 | | // If RouteConfiguration has been initialized, run the callback to continue in filter chain, |
208 | | // otherwise cache it and wait for the route table to be initialized. If RouteConfiguration hasn't |
209 | | // been initialized, routeConfig() return a shared_ptr to NullConfigImpl. The name of |
210 | | // NullConfigImpl is an empty string. |
211 | 0 | if (route_provider_ != nullptr && !routeConfig()->name().empty()) { |
212 | 0 | callback(); |
213 | 0 | return; |
214 | 0 | } |
215 | 0 | on_demand_update_callbacks_.push_back(callback); |
216 | | // Initialize the rds provider if it has not been initialized. There is potential race here |
217 | | // because other worker threads may also post callback to on demand update the RouteConfiguration |
218 | | // associated with this scope. If rds provider has been initialized, just wait for |
219 | | // RouteConfiguration to be updated. |
220 | 0 | maybeInitRdsConfigProvider(); |
221 | 0 | } |
222 | | |
223 | 0 | void ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper::runOnDemandUpdateCallback() { |
224 | 0 | for (auto& callback : on_demand_update_callbacks_) { |
225 | 0 | callback(); |
226 | 0 | } |
227 | 0 | on_demand_update_callbacks_.clear(); |
228 | 0 | } |
229 | | |
230 | | void ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper::initRdsConfigProvider( |
231 | | envoy::extensions::filters::network::http_connection_manager::v3::Rds& rds, |
232 | 0 | Init::Manager& init_manager) { |
233 | 0 | route_provider_ = std::dynamic_pointer_cast<RdsRouteConfigProviderImpl>( |
234 | 0 | parent_.route_config_provider_manager_.createRdsRouteConfigProvider( |
235 | 0 | rds, parent_.optional_http_filters_, parent_.factory_context_, parent_.stat_prefix_, |
236 | 0 | init_manager)); |
237 | |
|
238 | 0 | rds_update_callback_handle_ = route_provider_->subscription().addUpdateCallback([this]() { |
239 | | // Subscribe to RDS update. |
240 | 0 | parent_.onRdsConfigUpdate(scope_name_, route_provider_->configCast()); |
241 | 0 | }); |
242 | 0 | parent_.stats_.active_scopes_.inc(); |
243 | 0 | } |
244 | | |
245 | 0 | void ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper::maybeInitRdsConfigProvider() { |
246 | | // If the route provider have been initialized, return and wait for rds config update. |
247 | 0 | if (route_provider_ != nullptr) { |
248 | 0 | return; |
249 | 0 | } |
250 | | |
251 | | // Create a init_manager to create a rds provider. |
252 | | // No transitive warming dependency here because only on demand update reach this point. |
253 | 0 | Init::ManagerImpl srds_init_mgr("SRDS on demand init manager."); |
254 | 0 | Cleanup srds_initialization_continuation([this, &srds_init_mgr] { |
255 | 0 | Init::WatcherImpl noop_watcher( |
256 | 0 | fmt::format("SRDS on demand ConfigUpdate watcher: {}", scope_name_), |
257 | 0 | []() { /*Do nothing.*/ }); |
258 | 0 | srds_init_mgr.initialize(noop_watcher); |
259 | 0 | }); |
260 | | // Create route provider. |
261 | 0 | envoy::extensions::filters::network::http_connection_manager::v3::Rds rds; |
262 | 0 | rds.mutable_config_source()->MergeFrom(parent_.rds_config_source_); |
263 | 0 | rds.set_route_config_name( |
264 | 0 | parent_.scoped_route_map_[scope_name_]->configProto().route_configuration_name()); |
265 | 0 | initRdsConfigProvider(rds, srds_init_mgr); |
266 | 0 | ENVOY_LOG(debug, fmt::format("Scope on demand update: {}", scope_name_)); |
267 | | // If RouteConfiguration hasn't been initialized, routeConfig() return a shared_ptr to |
268 | | // NullConfigImpl. The name of NullConfigImpl is an empty string. |
269 | 0 | if (routeConfig()->name().empty()) { |
270 | 0 | return; |
271 | 0 | } |
272 | | // If RouteConfiguration has been initialized, apply update to all the threads. |
273 | 0 | parent_.onRdsConfigUpdate(scope_name_, route_provider_->configCast()); |
274 | 0 | } |
275 | | |
276 | | absl::StatusOr<bool> ScopedRdsConfigSubscription::addOrUpdateScopes( |
277 | | const std::vector<Envoy::Config::DecodedResourceRef>& resources, Init::Manager& init_manager, |
278 | 0 | const std::string& version_info) { |
279 | 0 | bool any_applied = false; |
280 | 0 | envoy::extensions::filters::network::http_connection_manager::v3::Rds rds; |
281 | 0 | rds.mutable_config_source()->MergeFrom(rds_config_source_); |
282 | 0 | std::vector<ScopedRouteInfoConstSharedPtr> updated_scopes; |
283 | 0 | for (const auto& resource : resources) { |
284 | | // Explicit copy so that we can std::move later. |
285 | 0 | envoy::config::route::v3::ScopedRouteConfiguration scoped_route_config = |
286 | 0 | dynamic_cast<const envoy::config::route::v3::ScopedRouteConfiguration&>( |
287 | 0 | resource.get().resource()); |
288 | 0 | if (scoped_route_config.route_configuration_name().empty()) { |
289 | 0 | return absl::InvalidArgumentError("route_configuration_name is empty."); |
290 | 0 | } |
291 | 0 | const std::string scope_name = scoped_route_config.name(); |
292 | 0 | if (const auto& scope_info_iter = scoped_route_map_.find(scope_name); |
293 | 0 | scope_info_iter != scoped_route_map_.end() && |
294 | 0 | scope_info_iter->second->configHash() == MessageUtil::hash(scoped_route_config)) { |
295 | 0 | continue; |
296 | 0 | } |
297 | 0 | rds.set_route_config_name(scoped_route_config.route_configuration_name()); |
298 | 0 | std::unique_ptr<RdsRouteConfigProviderHelper> rds_config_provider_helper; |
299 | 0 | std::shared_ptr<ScopedRouteInfo> scoped_route_info = nullptr; |
300 | 0 | if (scoped_route_config.on_demand() == false) { |
301 | | // For default scopes, create a rds helper with rds provider initialized. |
302 | 0 | rds_config_provider_helper = |
303 | 0 | std::make_unique<RdsRouteConfigProviderHelper>(*this, scope_name, rds, init_manager); |
304 | 0 | scoped_route_info = std::make_shared<ScopedRouteInfo>( |
305 | 0 | std::move(scoped_route_config), rds_config_provider_helper->routeConfig()); |
306 | 0 | } else { |
307 | | // For on demand scopes, create a rds helper with rds provider uninitialized. |
308 | 0 | rds_config_provider_helper = |
309 | 0 | std::make_unique<RdsRouteConfigProviderHelper>(*this, scope_name); |
310 | | // scope_route_info->routeConfig() will be nullptr, because RouteConfiguration is not loaded. |
311 | 0 | scoped_route_info = |
312 | 0 | std::make_shared<ScopedRouteInfo>(std::move(scoped_route_config), nullptr); |
313 | 0 | } |
314 | 0 | route_provider_by_scope_[scope_name] = std::move(rds_config_provider_helper); |
315 | 0 | scope_name_by_hash_[scoped_route_info->scopeKey().hash()] = scoped_route_info->scopeName(); |
316 | 0 | scoped_route_map_[scoped_route_info->scopeName()] = scoped_route_info; |
317 | 0 | updated_scopes.push_back(scoped_route_info); |
318 | 0 | any_applied = true; |
319 | 0 | ENVOY_LOG(debug, "srds: queueing add/update of scoped_route '{}', version: {}", |
320 | 0 | scoped_route_info->scopeName(), version_info); |
321 | 0 | } |
322 | | |
323 | | // scoped_route_info of both eager loading and on demand scopes will be propagated to work |
324 | | // threads. Upon a scoped RouteConfiguration miss, if the scope exists, an on demand update |
325 | | // callback will be posted to main thread. |
326 | 0 | if (!updated_scopes.empty()) { |
327 | 0 | applyConfigUpdate([updated_scopes](ConfigProvider::ConfigConstSharedPtr config) |
328 | 0 | -> ConfigProvider::ConfigConstSharedPtr { |
329 | 0 | auto* thread_local_scoped_config = |
330 | 0 | const_cast<ScopedConfigImpl*>(static_cast<const ScopedConfigImpl*>(config.get())); |
331 | 0 | thread_local_scoped_config->addOrUpdateRoutingScopes(updated_scopes); |
332 | 0 | return config; |
333 | 0 | }); |
334 | 0 | } |
335 | 0 | return any_applied; |
336 | 0 | } |
337 | | |
338 | | std::list<ScopedRdsConfigSubscription::RdsRouteConfigProviderHelperPtr> |
339 | | ScopedRdsConfigSubscription::removeScopes( |
340 | 0 | const Protobuf::RepeatedPtrField<std::string>& scope_names, const std::string& version_info) { |
341 | 0 | std::list<ScopedRdsConfigSubscription::RdsRouteConfigProviderHelperPtr> |
342 | 0 | to_be_removed_rds_providers; |
343 | 0 | std::vector<std::string> removed_scope_names; |
344 | 0 | for (const auto& scope_name : scope_names) { |
345 | 0 | auto iter = scoped_route_map_.find(scope_name); |
346 | 0 | if (iter != scoped_route_map_.end()) { |
347 | 0 | auto rds_config_provider_helper_iter = route_provider_by_scope_.find(scope_name); |
348 | 0 | if (rds_config_provider_helper_iter != route_provider_by_scope_.end()) { |
349 | 0 | to_be_removed_rds_providers.emplace_back( |
350 | 0 | std::move(rds_config_provider_helper_iter->second)); |
351 | 0 | route_provider_by_scope_.erase(rds_config_provider_helper_iter); |
352 | 0 | } |
353 | 0 | ASSERT(scope_name_by_hash_.find(iter->second->scopeKey().hash()) != |
354 | 0 | scope_name_by_hash_.end()); |
355 | 0 | scope_name_by_hash_.erase(iter->second->scopeKey().hash()); |
356 | 0 | scoped_route_map_.erase(iter); |
357 | 0 | removed_scope_names.push_back(scope_name); |
358 | 0 | ENVOY_LOG(debug, "srds: queueing removal of scoped route '{}', version: {}", scope_name, |
359 | 0 | version_info); |
360 | 0 | } |
361 | 0 | } |
362 | 0 | if (!removed_scope_names.empty()) { |
363 | 0 | applyConfigUpdate([removed_scope_names](ConfigProvider::ConfigConstSharedPtr config) |
364 | 0 | -> ConfigProvider::ConfigConstSharedPtr { |
365 | 0 | auto* thread_local_scoped_config = |
366 | 0 | const_cast<ScopedConfigImpl*>(static_cast<const ScopedConfigImpl*>(config.get())); |
367 | 0 | thread_local_scoped_config->removeRoutingScopes(removed_scope_names); |
368 | 0 | return config; |
369 | 0 | }); |
370 | 0 | } |
371 | 0 | return to_be_removed_rds_providers; |
372 | 0 | } |
373 | | |
374 | | absl::Status ScopedRdsConfigSubscription::onConfigUpdate( |
375 | | const std::vector<Envoy::Config::DecodedResourceRef>& added_resources, |
376 | | const Protobuf::RepeatedPtrField<std::string>& removed_resources, |
377 | 0 | const std::string& version_info) { |
378 | | // NOTE: deletes are done before adds/updates. |
379 | 0 | absl::flat_hash_map<std::string, ScopedRouteInfoConstSharedPtr> to_be_removed_scopes; |
380 | | // Destruction of resume_rds will lift the floodgate for new RDS subscriptions. |
381 | | // Note in the case of partial acceptance, accepted RDS subscriptions should be started |
382 | | // despite of any error. |
383 | 0 | ScopedResume resume_rds; |
384 | | // If new route config sources come after the local init manager's initialize() been |
385 | | // called, the init manager can't accept new targets. Instead we use a local override which will |
386 | | // start new subscriptions but not wait on them to be ready. |
387 | 0 | std::unique_ptr<Init::ManagerImpl> srds_init_mgr; |
388 | | // NOTE: This should be defined after srds_init_mgr and resume_rds, as it depends on the |
389 | | // srds_init_mgr, and we want a single RDS discovery request to be sent to management |
390 | | // server. |
391 | 0 | std::unique_ptr<Cleanup> srds_initialization_continuation; |
392 | 0 | ASSERT(localInitManager().state() > Init::Manager::State::Uninitialized); |
393 | 0 | const auto type_url = Envoy::Config::getTypeUrl<envoy::config::route::v3::RouteConfiguration>(); |
394 | | // Pause RDS to not send a burst of RDS requests until we start all the new subscriptions. |
395 | | // In the case that localInitManager is uninitialized, RDS is already paused |
396 | | // either by Server init or LDS init. |
397 | 0 | if (factory_context_.clusterManager().adsMux()) { |
398 | 0 | resume_rds = factory_context_.clusterManager().adsMux()->pause(type_url); |
399 | 0 | } |
400 | | // if local init manager is initialized, the parent init manager may have gone away. |
401 | 0 | if (localInitManager().state() == Init::Manager::State::Initialized) { |
402 | 0 | srds_init_mgr = |
403 | 0 | std::make_unique<Init::ManagerImpl>(fmt::format("SRDS {}:{}", name_, version_info)); |
404 | 0 | srds_initialization_continuation = |
405 | 0 | std::make_unique<Cleanup>([this, &srds_init_mgr, version_info] { |
406 | | // For new RDS subscriptions created after listener warming up, we don't wait for them to |
407 | | // warm up. |
408 | 0 | Init::WatcherImpl noop_watcher( |
409 | | // Note: we just throw it away. |
410 | 0 | fmt::format("SRDS ConfigUpdate watcher {}:{}", name_, version_info), |
411 | 0 | []() { /*Do nothing.*/ }); |
412 | 0 | srds_init_mgr->initialize(noop_watcher); |
413 | 0 | }); |
414 | 0 | } |
415 | |
|
416 | 0 | std::string exception_msg; |
417 | 0 | Protobuf::RepeatedPtrField<std::string> clean_removed_resources = |
418 | 0 | detectUpdateConflictAndCleanupRemoved(added_resources, removed_resources, exception_msg); |
419 | 0 | if (!exception_msg.empty()) { |
420 | 0 | return absl::InvalidArgumentError( |
421 | 0 | fmt::format("Error adding/updating scoped route(s): {}", exception_msg)); |
422 | 0 | } |
423 | | |
424 | | // Do not delete RDS config providers just yet, in case the to be deleted RDS subscriptions could |
425 | | // be reused by some to be added scopes. |
426 | 0 | std::list<ScopedRdsConfigSubscription::RdsRouteConfigProviderHelperPtr> |
427 | 0 | to_be_removed_rds_providers = removeScopes(clean_removed_resources, version_info); |
428 | |
|
429 | 0 | auto status_or_applied = addOrUpdateScopes( |
430 | 0 | added_resources, (srds_init_mgr == nullptr ? localInitManager() : *srds_init_mgr), |
431 | 0 | version_info); |
432 | 0 | if (!status_or_applied.status().ok()) { |
433 | 0 | return status_or_applied.status(); |
434 | 0 | } |
435 | 0 | const bool any_applied = status_or_applied.value(); |
436 | 0 | const auto status = ConfigSubscriptionCommonBase::onConfigUpdate(); |
437 | 0 | if (!status.ok()) { |
438 | 0 | return status; |
439 | 0 | } |
440 | 0 | if (any_applied || !to_be_removed_rds_providers.empty()) { |
441 | 0 | setLastConfigInfo(absl::optional<LastConfigInfo>({absl::nullopt, version_info})); |
442 | 0 | } |
443 | 0 | stats_.all_scopes_.set(scoped_route_map_.size()); |
444 | 0 | stats_.config_reload_.inc(); |
445 | 0 | stats_.config_reload_time_ms_.set(DateUtil::nowToMilliseconds(factory_context_.timeSource())); |
446 | 0 | return absl::OkStatus(); |
447 | 0 | } |
448 | | |
449 | | void ScopedRdsConfigSubscription::onRdsConfigUpdate(const std::string& scope_name, |
450 | 0 | ConfigConstSharedPtr new_rds_config) { |
451 | 0 | auto iter = scoped_route_map_.find(scope_name); |
452 | 0 | ASSERT(iter != scoped_route_map_.end(), |
453 | 0 | fmt::format("trying to update route config for non-existing scope {}", scope_name)); |
454 | 0 | auto new_scoped_route_info = std::make_shared<ScopedRouteInfo>( |
455 | 0 | envoy::config::route::v3::ScopedRouteConfiguration(iter->second->configProto()), |
456 | 0 | std::move(new_rds_config)); |
457 | 0 | scoped_route_map_[new_scoped_route_info->scopeName()] = new_scoped_route_info; |
458 | 0 | applyConfigUpdate([new_scoped_route_info](ConfigProvider::ConfigConstSharedPtr config) |
459 | 0 | -> ConfigProvider::ConfigConstSharedPtr { |
460 | 0 | auto* thread_local_scoped_config = |
461 | 0 | const_cast<ScopedConfigImpl*>(static_cast<const ScopedConfigImpl*>(config.get())); |
462 | 0 | thread_local_scoped_config->addOrUpdateRoutingScopes({new_scoped_route_info}); |
463 | 0 | return config; |
464 | 0 | }); |
465 | | // The data plane may wait for the route configuration to come back. |
466 | 0 | route_provider_by_scope_[scope_name]->runOnDemandUpdateCallback(); |
467 | 0 | } |
468 | | |
469 | | // TODO(stevenzzzz): see issue #7508, consider generalizing this function as it overlaps with |
470 | | // CdsApiImpl::onConfigUpdate. |
471 | | absl::Status ScopedRdsConfigSubscription::onConfigUpdate( |
472 | | const std::vector<Envoy::Config::DecodedResourceRef>& resources, |
473 | 0 | const std::string& version_info) { |
474 | 0 | Protobuf::RepeatedPtrField<std::string> to_remove_repeated; |
475 | 0 | for (const auto& scoped_route : scoped_route_map_) { |
476 | 0 | *to_remove_repeated.Add() = scoped_route.first; |
477 | 0 | } |
478 | 0 | return onConfigUpdate(resources, to_remove_repeated, version_info); |
479 | 0 | } |
480 | | |
481 | | Protobuf::RepeatedPtrField<std::string> |
482 | | ScopedRdsConfigSubscription::detectUpdateConflictAndCleanupRemoved( |
483 | | const std::vector<Envoy::Config::DecodedResourceRef>& resources, |
484 | 0 | const Protobuf::RepeatedPtrField<std::string>& removed_resources, std::string& exception_msg) { |
485 | 0 | Protobuf::RepeatedPtrField<std::string> clean_removed_resources; |
486 | | // All the scope names to be removed or updated. |
487 | 0 | absl::flat_hash_set<std::string> updated_or_removed_scopes; |
488 | 0 | for (const std::string& removed_resource : removed_resources) { |
489 | 0 | updated_or_removed_scopes.insert(removed_resource); |
490 | 0 | } |
491 | 0 | for (const auto& resource : resources) { |
492 | 0 | const auto& scoped_route = |
493 | 0 | dynamic_cast<const envoy::config::route::v3::ScopedRouteConfiguration&>( |
494 | 0 | resource.get().resource()); |
495 | 0 | updated_or_removed_scopes.insert(scoped_route.name()); |
496 | 0 | } |
497 | |
|
498 | 0 | absl::flat_hash_map<uint64_t, std::string> scope_name_by_hash = scope_name_by_hash_; |
499 | 0 | absl::erase_if(scope_name_by_hash, [&updated_or_removed_scopes](const auto& key_name) { |
500 | 0 | auto const& [key, name] = key_name; |
501 | 0 | UNREFERENCED_PARAMETER(key); |
502 | 0 | return updated_or_removed_scopes.contains(name); |
503 | 0 | }); |
504 | 0 | absl::flat_hash_map<std::string, envoy::config::route::v3::ScopedRouteConfiguration> |
505 | 0 | scoped_routes; |
506 | 0 | for (const auto& resource : resources) { |
507 | | // Throws (thus rejects all) on any error. |
508 | 0 | const auto& scoped_route = |
509 | 0 | dynamic_cast<const envoy::config::route::v3::ScopedRouteConfiguration&>( |
510 | 0 | resource.get().resource()); |
511 | 0 | const std::string& scope_name = scoped_route.name(); |
512 | 0 | auto scope_config_inserted = scoped_routes.try_emplace(scope_name, std::move(scoped_route)); |
513 | 0 | if (!scope_config_inserted.second) { |
514 | 0 | exception_msg = fmt::format("duplicate scoped route configuration '{}' found", scope_name); |
515 | 0 | return clean_removed_resources; |
516 | 0 | } |
517 | 0 | envoy::config::route::v3::ScopedRouteConfiguration scoped_route_config = |
518 | 0 | scope_config_inserted.first->second; |
519 | 0 | const uint64_t key_fingerprint = |
520 | 0 | ScopedRouteInfo(std::move(scoped_route_config), nullptr).scopeKey().hash(); |
521 | 0 | if (!scope_name_by_hash.try_emplace(key_fingerprint, scope_name).second) { |
522 | 0 | exception_msg = |
523 | 0 | fmt::format("scope key conflict found, first scope is '{}', second scope is '{}'", |
524 | 0 | scope_name_by_hash[key_fingerprint], scope_name); |
525 | 0 | return clean_removed_resources; |
526 | 0 | } |
527 | 0 | } |
528 | | |
529 | | // only remove resources that is not going to be updated. |
530 | 0 | for (const std::string& removed_resource : removed_resources) { |
531 | 0 | if (!scoped_routes.contains(removed_resource)) { |
532 | 0 | *clean_removed_resources.Add() = removed_resource; |
533 | 0 | } |
534 | 0 | } |
535 | 0 | return clean_removed_resources; |
536 | 0 | } |
537 | | |
538 | | void ScopedRdsConfigSubscription::onDemandRdsUpdate( |
539 | | std::shared_ptr<Router::ScopeKey> scope_key, Event::Dispatcher& thread_local_dispatcher, |
540 | | Http::RouteConfigUpdatedCallback&& route_config_updated_cb, |
541 | 0 | std::weak_ptr<Envoy::Config::ConfigSubscriptionCommonBase> weak_subscription) { |
542 | 0 | factory_context_.mainThreadDispatcher().post([this, &thread_local_dispatcher, scope_key, |
543 | 0 | route_config_updated_cb, weak_subscription]() { |
544 | | // If the subscription has been destroyed, return immediately. |
545 | 0 | if (!weak_subscription.lock()) { |
546 | 0 | thread_local_dispatcher.post([route_config_updated_cb] { route_config_updated_cb(false); }); |
547 | 0 | return; |
548 | 0 | } |
549 | | |
550 | 0 | auto iter = scope_name_by_hash_.find(scope_key->hash()); |
551 | | // Return to filter chain if we can't find the scope. |
552 | | // The scope may have been destroyed when callback reach the main thread. |
553 | 0 | if (iter == scope_name_by_hash_.end()) { |
554 | 0 | thread_local_dispatcher.post([route_config_updated_cb] { route_config_updated_cb(false); }); |
555 | 0 | return; |
556 | 0 | } |
557 | | // Wrap the thread local dispatcher inside the callback. |
558 | 0 | std::function<void()> thread_local_updated_callback = [route_config_updated_cb, |
559 | 0 | &thread_local_dispatcher]() { |
560 | 0 | thread_local_dispatcher.post([route_config_updated_cb] { route_config_updated_cb(true); }); |
561 | 0 | }; |
562 | 0 | std::string scope_name = iter->second; |
563 | | // On demand initialization inside main thread. |
564 | 0 | route_provider_by_scope_[scope_name]->addOnDemandUpdateCallback(thread_local_updated_callback); |
565 | 0 | }); |
566 | 0 | } |
567 | | |
568 | | ScopedRdsConfigProvider::ScopedRdsConfigProvider( |
569 | | ScopedRdsConfigSubscriptionSharedPtr&& subscription) |
570 | 8 | : MutableConfigProviderCommonBase(std::move(subscription), ConfigProvider::ApiType::Delta) {} |
571 | | |
572 | | ProtobufTypes::MessagePtr |
573 | 2.63k | ScopedRoutesConfigProviderManager::dumpConfigs(const Matchers::StringMatcher& name_matcher) const { |
574 | 2.63k | auto config_dump = std::make_unique<envoy::admin::v3::ScopedRoutesConfigDump>(); |
575 | 2.63k | for (const auto& element : configSubscriptions()) { |
576 | 0 | auto subscription = element.second.lock(); |
577 | 0 | ASSERT(subscription); |
578 | | |
579 | 0 | if (subscription->configInfo()) { |
580 | 0 | auto* dynamic_config = config_dump->mutable_dynamic_scoped_route_configs()->Add(); |
581 | 0 | dynamic_config->set_version_info(subscription->configInfo().value().last_config_version_); |
582 | 0 | const ScopedRdsConfigSubscription* typed_subscription = |
583 | 0 | static_cast<ScopedRdsConfigSubscription*>(subscription.get()); |
584 | 0 | dynamic_config->set_name(typed_subscription->name()); |
585 | 0 | const ScopedRouteMap& scoped_route_map = typed_subscription->scopedRouteMap(); |
586 | 0 | for (const auto& it : scoped_route_map) { |
587 | 0 | if (!name_matcher.match(it.second->configProto().name())) { |
588 | 0 | continue; |
589 | 0 | } |
590 | 0 | dynamic_config->mutable_scoped_route_configs()->Add()->PackFrom(it.second->configProto()); |
591 | 0 | } |
592 | 0 | TimestampUtil::systemClockToTimestamp(subscription->lastUpdated(), |
593 | 0 | *dynamic_config->mutable_last_updated()); |
594 | 0 | } |
595 | 0 | } |
596 | | |
597 | 2.63k | for (const auto& provider : immutableConfigProviders(ConfigProviderInstanceType::Inline)) { |
598 | 0 | const auto protos_info = |
599 | 0 | provider->configProtoInfoVector<envoy::config::route::v3::ScopedRouteConfiguration>(); |
600 | 0 | ASSERT(protos_info != absl::nullopt); |
601 | 0 | auto* inline_config = config_dump->mutable_inline_scoped_route_configs()->Add(); |
602 | 0 | inline_config->set_name(static_cast<InlineScopedRoutesConfigProvider*>(provider)->name()); |
603 | 0 | for (const auto& config_proto : protos_info.value().config_protos_) { |
604 | 0 | if (!name_matcher.match(config_proto->name())) { |
605 | 0 | continue; |
606 | 0 | } |
607 | 0 | inline_config->mutable_scoped_route_configs()->Add()->PackFrom(*config_proto); |
608 | 0 | } |
609 | 0 | TimestampUtil::systemClockToTimestamp(provider->lastUpdated(), |
610 | 0 | *inline_config->mutable_last_updated()); |
611 | 0 | } |
612 | | |
613 | 2.63k | return config_dump; |
614 | 2.63k | } |
615 | | |
616 | | ConfigProviderPtr ScopedRoutesConfigProviderManager::createXdsConfigProvider( |
617 | | const Protobuf::Message& config_source_proto, |
618 | | Server::Configuration::ServerFactoryContext& factory_context, Init::Manager& init_manager, |
619 | 9 | const std::string& stat_prefix, const ConfigProviderManager::OptionalArg& optarg) { |
620 | 9 | const auto& typed_optarg = static_cast<const ScopedRoutesConfigProviderManagerOptArg&>(optarg); |
621 | 9 | ScopedRdsConfigSubscriptionSharedPtr subscription = |
622 | 9 | ConfigProviderManagerImplBase::getSubscription<ScopedRdsConfigSubscription>( |
623 | 9 | config_source_proto, init_manager, |
624 | 9 | [&config_source_proto, &factory_context, &stat_prefix, |
625 | 9 | &typed_optarg](const uint64_t manager_identifier, |
626 | 9 | ConfigProviderManagerImplBase& config_provider_manager) |
627 | 9 | -> Envoy::Config::ConfigSubscriptionCommonBaseSharedPtr { |
628 | 7 | const auto& scoped_rds_config_source = dynamic_cast< |
629 | 7 | const envoy::extensions::filters::network::http_connection_manager::v3::ScopedRds&>( |
630 | 7 | config_source_proto); |
631 | 7 | return std::make_shared<ScopedRdsConfigSubscription>( |
632 | 7 | scoped_rds_config_source, typed_optarg.optional_http_filters_, manager_identifier, |
633 | 7 | typed_optarg.scoped_routes_name_, factory_context, stat_prefix, |
634 | 7 | typed_optarg.rds_config_source_, |
635 | 7 | static_cast<ScopedRoutesConfigProviderManager&>(config_provider_manager) |
636 | 7 | .routeConfigProviderManager(), |
637 | 7 | static_cast<ScopedRoutesConfigProviderManager&>(config_provider_manager)); |
638 | 7 | }); |
639 | | |
640 | 9 | return std::make_unique<ScopedRdsConfigProvider>(std::move(subscription)); |
641 | 9 | } |
642 | | |
643 | | ConfigProviderPtr ScopedRoutesConfigProviderManager::createStaticConfigProvider( |
644 | | ProtobufTypes::ConstMessagePtrVector&& config_protos, |
645 | | Server::Configuration::ServerFactoryContext& factory_context, |
646 | 92 | const ConfigProviderManager::OptionalArg& optarg) { |
647 | 92 | const auto& typed_optarg = static_cast<const ScopedRoutesConfigProviderManagerOptArg&>(optarg); |
648 | 92 | return std::make_unique<InlineScopedRoutesConfigProvider>( |
649 | 92 | std::move(config_protos), typed_optarg.scoped_routes_name_, factory_context, *this, |
650 | 92 | typed_optarg.rds_config_source_, typed_optarg.optional_http_filters_); |
651 | 92 | } |
652 | | |
653 | | } // namespace Router |
654 | | } // namespace Envoy |