Grcov report - eds.cc

1

#include "source/extensions/clusters/eds/eds.h"

2

3

#include "envoy/common/exception.h"

4

#include "envoy/config/cluster/v3/cluster.pb.h"

5

#include "envoy/config/core/v3/config_source.pb.h"

6

#include "envoy/service/discovery/v3/discovery.pb.h"

7

8

#include "source/common/common/assert.h"

9

#include "source/common/common/utility.h"

10

#include "source/common/config/api_version.h"

11

#include "source/common/config/decoded_resource_impl.h"

12

#include "source/common/grpc/common.h"

13

#include "source/common/runtime/runtime_features.h"

14

15

namespace Envoy {

16

namespace Upstream {

17

18

absl::StatusOr<std::unique_ptr<EdsClusterImpl>>

19

EdsClusterImpl::create(const envoy::config::cluster::v3::Cluster& cluster,

20

612

                       ClusterFactoryContext& cluster_context) {

21

612

  absl::Status creation_status = absl::OkStatus();

22

612

  std::unique_ptr<EdsClusterImpl> ret =

23

612

      absl::WrapUnique(new EdsClusterImpl(cluster, cluster_context, creation_status));

24

612

  RETURN_IF_NOT_OK(creation_status);

25

611

  return ret;

26

612

27

28

EdsClusterImpl::EdsClusterImpl(const envoy::config::cluster::v3::Cluster& cluster,

29

                               ClusterFactoryContext& cluster_context,

30

                               absl::Status& creation_status)

31

612

    : BaseDynamicClusterImpl(cluster, cluster_context, creation_status),

32

612

      Envoy::Config::SubscriptionBase<envoy::config::endpoint::v3::ClusterLoadAssignment>(

33

612

          cluster_context.messageValidationVisitor(), "cluster_name"),

34

612

      local_info_(cluster_context.serverFactoryContext().localInfo()),

35

      eds_resources_cache_(

36

612

          cluster_context.serverFactoryContext().clusterManager().edsResourcesCache()) {

37

612

  RETURN_ONLY_IF_NOT_OK_REF(creation_status);

38

611

  Event::Dispatcher& dispatcher = cluster_context.serverFactoryContext().mainThreadDispatcher();

39

611

  assignment_timeout_ = dispatcher.createTimer([this]() -> void { onAssignmentTimeout(); });

40

611

  const auto& eds_config = cluster.eds_cluster_config().eds_config();

41

611

  if (Config::SubscriptionFactory::isPathBasedConfigSource(

42

611

          eds_config.config_source_specifier_case())) {

43

97

    initialize_phase_ = InitializePhase::Primary;

44

604

  } else {

45

514

    initialize_phase_ = InitializePhase::Secondary;

46

514

47

611

  const auto resource_name = getResourceName();

48

611

  if (Runtime::runtimeFeatureEnabled(

49

611

          "envoy.reloadable_features.xdstp_based_config_singleton_subscriptions")) {

50

42

    subscription_ = THROW_OR_RETURN_VALUE(

51

42

        cluster_context.serverFactoryContext().xdsManager().subscribeToSingletonResource(

52

42

            edsServiceName(), eds_config, Grpc::Common::typeUrl(resource_name), info_->statsScope(),

53

42

            *this, resource_decoder_, {}),

54

42

        Config::SubscriptionPtr);

55

609

  } else {

56

569

    subscription_ = THROW_OR_RETURN_VALUE(

57

569

        cluster_context.serverFactoryContext()

58

569

            .clusterManager()

59

569

            .subscriptionFactory()

60

569

            .subscriptionFromConfigSource(eds_config, Grpc::Common::typeUrl(resource_name),

61

569

                                          info_->statsScope(), *this, resource_decoder_, {}),

62

569

        Config::SubscriptionPtr);

63

569

64

611

65

66

606

EdsClusterImpl::~EdsClusterImpl() {

67

606

  if (using_cached_resource_) {

68

    // Clear the callback as the subscription is no longer valid.

69

10

    eds_resources_cache_->removeCallback(edsServiceName(), this);

70

10

71

606

72

73

551

void EdsClusterImpl::startPreInit() { subscription_->start({edsServiceName()}); }

74

75

800

void EdsClusterImpl::BatchUpdateHelper::batchUpdate(PrioritySet::HostUpdateCb& host_update_cb) {

76

800

  absl::flat_hash_set<std::string> all_new_hosts;

77

800

  PriorityStateManager priority_state_manager(parent_, parent_.local_info_, &host_update_cb);

78

1184

  for (const auto& locality_lb_endpoint : cluster_load_assignment_.endpoints()) {

79

1184

    priority_state_manager.initializePriorityFor(locality_lb_endpoint);

80

81

1184

    if (locality_lb_endpoint.has_leds_cluster_locality_config()) {

82

      // The locality uses LEDS, fetch its dynamic data, which must be ready, or otherwise

83

      // the batchUpdate method should not have been called.

84

77

      const auto& leds_config = locality_lb_endpoint.leds_cluster_locality_config();

85

86

      // The batchUpdate call must be performed after all the endpoints of all localities

87

      // were received.

88

77

      const auto it = parent_.leds_localities_.find(leds_config);

89

77

      ASSERT(it != parent_.leds_localities_.end() && it->second->isUpdated());

90

115

      for (const auto& [_, lb_endpoint] : it->second->getEndpointsMap()) {

91

114

        updateLocalityEndpoints(lb_endpoint, locality_lb_endpoint, priority_state_manager,

92

114

                                all_new_hosts);

93

114

94

1173

    } else {

95

1313

      for (const auto& lb_endpoint : locality_lb_endpoint.lb_endpoints()) {

96

1119

        updateLocalityEndpoints(lb_endpoint, locality_lb_endpoint, priority_state_manager,

97

1119

                                all_new_hosts);

98

1119

99

1107

100

1184

101

102

  // Track whether we rebuilt any LB structures.

103

800

  bool cluster_rebuilt = false;

104

105

  // Get the map of all the latest existing hosts, which is used to filter out the existing

106

  // hosts in the process of updating cluster memberships.

107

800

  HostMapConstSharedPtr all_hosts = parent_.prioritySet().crossPriorityHostMap();

108

800

  ASSERT(all_hosts != nullptr);

109

110

800

  const uint32_t overprovisioning_factor = PROTOBUF_GET_WRAPPED_OR_DEFAULT(

111

800

      cluster_load_assignment_.policy(), overprovisioning_factor, kDefaultOverProvisioningFactor);

112

800

  const bool weighted_priority_health =

113

800

      cluster_load_assignment_.policy().weighted_priority_health();

114

115

800

  LocalityWeightsMap empty_locality_map;

116

117

  // Loop over all priorities that exist in the new configuration.

118

800

  auto& priority_state = priority_state_manager.priorityState();

119

800

  THROW_IF_NOT_OK(parent_.validateEndpoints(cluster_load_assignment_.endpoints(), priority_state));

120

1712

  for (size_t i = 0; i < priority_state.size(); ++i) {

121

917

    if (parent_.locality_weights_map_.size() <= i) {

122

565

      parent_.locality_weights_map_.resize(i + 1);

123

565

124

917

    if (priority_state[i].first != nullptr) {

125

894

      cluster_rebuilt |= parent_.updateHostsPerLocality(

126

894

          i, weighted_priority_health, overprovisioning_factor, *priority_state[i].first,

127

894

          parent_.locality_weights_map_[i], priority_state[i].second, priority_state_manager,

128

894

          *all_hosts, all_new_hosts);

129

894

    } else {

130

      // If the new update contains a priority with no hosts, call the update function with an empty

131

      // set of hosts.

132

23

      cluster_rebuilt |=

133

23

          parent_.updateHostsPerLocality(i, weighted_priority_health, overprovisioning_factor, {},

134

23

                                         parent_.locality_weights_map_[i], empty_locality_map,

135

23

                                         priority_state_manager, *all_hosts, all_new_hosts);

136

23

137

917

138

139

  // Loop over all priorities not present in the config that already exists. This will

140

  // empty out any remaining priority that the config update did not refer to.

141

822

  for (size_t i = priority_state.size(); i < parent_.priority_set_.hostSetsPerPriority().size();

142

795

       ++i) {

143

27

    if (parent_.locality_weights_map_.size() <= i) {

144

6

      parent_.locality_weights_map_.resize(i + 1);

145

6

146

27

    cluster_rebuilt |= parent_.updateHostsPerLocality(

147

27

        i, weighted_priority_health, overprovisioning_factor, {}, parent_.locality_weights_map_[i],

148

27

        empty_locality_map, priority_state_manager, *all_hosts, all_new_hosts);

149

27

150

151

795

  if (!cluster_rebuilt) {

152

78

    parent_.info_->configUpdateStats().update_no_rebuild_.inc();

153

78

154

155

  // If we didn't setup to initialize when our first round of health checking is complete, just

156

  // do it now.

157

795

  parent_.onPreInitComplete();

158

795

159

160

void EdsClusterImpl::BatchUpdateHelper::updateLocalityEndpoints(

161

    const envoy::config::endpoint::v3::LbEndpoint& lb_endpoint,

162

    const envoy::config::endpoint::v3::LocalityLbEndpoints& locality_lb_endpoint,

163

1233

    PriorityStateManager& priority_state_manager, absl::flat_hash_set<std::string>& all_new_hosts) {

164

1233

  const auto address =

165

1233

      THROW_OR_RETURN_VALUE(parent_.resolveProtoAddress(lb_endpoint.endpoint().address()),

166

1233

                            const Network::Address::InstanceConstSharedPtr);

167

1233

  std::vector<Network::Address::InstanceConstSharedPtr> address_list;

168

1233

  if (!lb_endpoint.endpoint().additional_addresses().empty()) {

169

3

    address_list.push_back(address);

170

3

    for (const auto& additional_address : lb_endpoint.endpoint().additional_addresses()) {

171

3

      Network::Address::InstanceConstSharedPtr address =

172

3

          returnOrThrow(parent_.resolveProtoAddress(additional_address.address()));

173

3

      address_list.emplace_back(address);

174

3

175

3

    if (!Runtime::runtimeFeatureEnabled(

176

3

            "envoy.reloadable_features.happy_eyeballs_sort_non_ip_addresses")) {

177

2

      for (const Network::Address::InstanceConstSharedPtr& address : address_list) {

178

        // All addresses must by IP addresses.

179

2

        if (!address->ip()) {

180

1

          throwEnvoyExceptionOrPanic("additional_addresses must be IP addresses.");

181

1

182

2

183

1

184

3

185

186

  // When the configuration contains duplicate hosts, only the first one will be retained.

187

1232

  const auto address_as_string = address->asString();

188

1232

  if (all_new_hosts.contains(address_as_string)) {

189

16

    return;

190

16

191

192

1216

  priority_state_manager.registerHostForPriority(lb_endpoint.endpoint().hostname(), address,

193

1216

                                                 address_list, locality_lb_endpoint, lb_endpoint);

194

1216

  all_new_hosts.emplace(address_as_string);

195

1216

196

197

absl::Status

198

EdsClusterImpl::onConfigUpdate(const std::vector<Config::DecodedResourceRef>& resources,

199

785

                               const std::string&) {

200

785

  if (resources.empty()) {

201

29

    ENVOY_LOG(debug, "Missing ClusterLoadAssignment for {} in onConfigUpdate()", edsServiceName());

202

29

    info_->configUpdateStats().update_empty_.inc();

203

29

    onPreInitComplete();

204

29

    return absl::OkStatus();

205

29

206

756

  if (resources.size() != 1) {

207

1

    return absl::InvalidArgumentError(

208

1

        fmt::format("Unexpected EDS resource length: {}", resources.size()));

209

1

210

211

755

  envoy::config::endpoint::v3::ClusterLoadAssignment cluster_load_assignment =

212

755

      dynamic_cast<const envoy::config::endpoint::v3::ClusterLoadAssignment&>(

213

755

          resources[0].get().resource());

214

755

  if (cluster_load_assignment.cluster_name() != edsServiceName()) {

215

1

    return absl::InvalidArgumentError(fmt::format("Unexpected EDS cluster (expecting {}): {}",

216

1

                                                  edsServiceName(),

217

1

                                                  cluster_load_assignment.cluster_name()));

218

1

219

  // Validate that each locality doesn't have both LEDS and endpoints defined.

220

  // TODO(adisuissa): This is only needed for the API v3 support. In future major versions

221

  // the oneof definition will take care of it.

222

1129

  for (const auto& locality : cluster_load_assignment.endpoints()) {

223

1129

    if (locality.has_leds_cluster_locality_config() && locality.lb_endpoints_size() > 0) {

224

1

      return absl::InvalidArgumentError(fmt::format(

225

1

          "A ClusterLoadAssignment for cluster {} cannot include both LEDS (resource: {}) and a "

226

1

          "list of endpoints.",

227

1

          edsServiceName(), locality.leds_cluster_locality_config().leds_collection_name()));

228

1

229

1129

230

231

  // Disable timer (if enabled) as we have received new assignment.

232

753

  if (assignment_timeout_->enabled()) {

233

2

    assignment_timeout_->disableTimer();

234

2

    if (eds_resources_cache_.has_value()) {

235

      eds_resources_cache_->disableExpiryTimer(edsServiceName());

236

237

2

238

  // Check if endpoint_stale_after is set.

239

753

  const uint64_t stale_after_ms =

240

753

      PROTOBUF_GET_MS_OR_DEFAULT(cluster_load_assignment.policy(), endpoint_stale_after, 0);

241

753

  if (stale_after_ms > 0) {

242

    // Stat to track how often we receive valid assignment_timeout in response.

243

8

    info_->configUpdateStats().assignment_timeout_received_.inc();

244

8

    assignment_timeout_->enableTimer(std::chrono::milliseconds(stale_after_ms));

245

8

    if (eds_resources_cache_.has_value()) {

246

3

      eds_resources_cache_->setExpiryTimer(edsServiceName(),

247

3

                                           std::chrono::milliseconds(stale_after_ms));

248

3

249

8

250

251

  // Pause LEDS messages until the EDS config is finished processing.

252

753

  const auto type_url = Config::getTypeUrl<envoy::config::endpoint::v3::LbEndpoint>();

253

753

  Config::ScopedResume resume_leds =

254

753

      transport_factory_context_->serverFactoryContext().xdsManager().pause(type_url);

255

256

753

  update(std::move(cluster_load_assignment));

257

  // If previously used a cached version, remove the subscription from the cache's

258

  // callbacks.

259

753

  if (using_cached_resource_) {

260

9

    eds_resources_cache_->removeCallback(edsServiceName(), this);

261

9

    using_cached_resource_ = false;

262

9

263

753

  return absl::OkStatus();

264

754

265

266

void EdsClusterImpl::update(

267

775

    envoy::config::endpoint::v3::ClusterLoadAssignment&& cluster_load_assignment) {

268

  // Drop overload configuration parsing.

269

775

  THROW_IF_NOT_OK(parseDropOverloadConfig(cluster_load_assignment));

270

271

  // Compare the current set of LEDS localities (localities using LEDS) to the one received in the

272

  // update. A LEDS locality can either be added, removed, or kept. If it is added we add a

273

  // subscription to it, and if it is removed we delete the subscription.

274

775

  LedsConfigSet cla_leds_configs;

275

276

1147

  for (const auto& locality : cluster_load_assignment.endpoints()) {

277

1147

    if (locality.has_leds_cluster_locality_config()) {

278

40

      cla_leds_configs.emplace(locality.leds_cluster_locality_config());

279

40

280

1147

281

282

  // Remove the LEDS localities that are not needed anymore.

283

775

  absl::erase_if(leds_localities_, [&cla_leds_configs](const auto& item) {

284

5

    auto const& [leds_config, _] = item;

285

    // Returns true if the leds_config isn't in the cla_leds_configs

286

5

    return cla_leds_configs.find(leds_config) == cla_leds_configs.end();

287

5

});

288

289

775

  const envoy::config::endpoint::v3::ClusterLoadAssignment* used_load_assignment;

290

775

  if (!cla_leds_configs.empty() || eds_resources_cache_.has_value()) {

291

388

    cluster_load_assignment_ = std::make_unique<envoy::config::endpoint::v3::ClusterLoadAssignment>(

292

388

        std::move(cluster_load_assignment));

293

388

    used_load_assignment = cluster_load_assignment_.get();

294

758

  } else {

295

387

    cluster_load_assignment_ = nullptr;

296

387

    used_load_assignment = &cluster_load_assignment;

297

387

298

299

  // Add all the LEDS localities that are new.

300

779

  for (const auto& leds_config : cla_leds_configs) {

301

40

    if (leds_localities_.find(leds_config) == leds_localities_.end()) {

302

39

      ENVOY_LOG(trace, "Found new LEDS config in EDS onConfigUpdate() for cluster {}: {}",

303

39

                edsServiceName(), leds_config.DebugString());

304

305

      // Create a new LEDS subscription and add it to the subscriptions map.

306

39

      LedsSubscriptionPtr leds_locality_subscription = std::make_unique<LedsSubscription>(

307

39

          leds_config, edsServiceName(), *transport_factory_context_, info_->statsScope(),

308

69

          [this]() {

309

65

            if (validateAllLedsUpdated()) {

310

61

              ASSERT(cluster_load_assignment_ != nullptr);

311

61

              BatchUpdateHelper helper(*this, *cluster_load_assignment_);

312

61

              priority_set_.batchHostUpdate(helper);

313

61

314

65

});

315

39

      leds_localities_.emplace(leds_config, std::move(leds_locality_subscription));

316

39

317

40

318

319

  // If all the LEDS localities are updated, the EDS update can occur. If not, then when the last

320

  // LEDS locality will be updated, it will trigger the EDS update helper.

321

775

  if (!validateAllLedsUpdated()) {

322

36

    return;

323

36

324

325

739

  BatchUpdateHelper helper(*this, *used_load_assignment);

326

739

  priority_set_.batchHostUpdate(helper);

327

739

328

329

absl::Status

330

EdsClusterImpl::onConfigUpdate(const std::vector<Config::DecodedResourceRef>& added_resources,

331

233

                               const Protobuf::RepeatedPtrField<std::string>&, const std::string&) {

332

233

  return onConfigUpdate(added_resources, "");

333

233

334

335

2

void EdsClusterImpl::onAssignmentTimeout() {

336

  // We can no longer use the assignments, remove them.

337

  // TODO(vishalpowar) This is not going to work for incremental updates, and we

338

  // need to instead change the health status to indicate the assignments are

339

  // stale.

340

  // TODO(snowp): This should probably just use xDS TTLs?

341

2

  envoy::config::endpoint::v3::ClusterLoadAssignment resource;

342

2

  resource.set_cluster_name(edsServiceName());

343

2

  update(std::move(resource));

344

345

2

  if (eds_resources_cache_.has_value()) {

346

    // Clear the resource so it won't be used, and its watchers will be notified.

347

1

    eds_resources_cache_->removeResource(edsServiceName());

348

1

349

  // Stat to track how often we end up with stale assignments.

350

2

  info_->configUpdateStats().assignment_stale_.inc();

351

2

352

353

1

void EdsClusterImpl::onCachedResourceRemoved(absl::string_view resource_name) {

354

1

  ASSERT(resource_name == edsServiceName());

355

  // Disable the timer if previously started.

356

1

  if (assignment_timeout_->enabled()) {

357

    assignment_timeout_->disableTimer();

358

    eds_resources_cache_->disableExpiryTimer(edsServiceName());

359

360

1

  envoy::config::endpoint::v3::ClusterLoadAssignment resource;

361

1

  resource.set_cluster_name(edsServiceName());

362

1

  update(std::move(resource));

363

1

364

365

66

void EdsClusterImpl::reloadHealthyHostsHelper(const HostSharedPtr& host) {

366

  // Here we will see if we have a host that has been marked for deletion by service discovery

367

  // but has been stabilized due to passing active health checking. If such a host is now

368

  // failing active health checking we can remove it during this health check update.

369

66

  HostSharedPtr host_to_exclude = host;

370

66

  if (host_to_exclude != nullptr &&

371

66

      host_to_exclude->healthFlagGet(Host::HealthFlag::FAILED_ACTIVE_HC) &&

372

66

      host_to_exclude->healthFlagGet(Host::HealthFlag::PENDING_DYNAMIC_REMOVAL)) {

373

    // Empty for clarity.

374

61

  } else {

375

    // Do not exclude and remove the host during the update.

376

61

    host_to_exclude = nullptr;

377

61

378

379

66

  const auto& host_sets = prioritySet().hostSetsPerPriority();

380

132

  for (size_t priority = 0; priority < host_sets.size(); ++priority) {

381

66

    const auto& host_set = host_sets[priority];

382

383

    // Filter current hosts in case we need to exclude a host.

384

66

    HostVectorSharedPtr hosts_copy(new HostVector());

385

66

    std::copy_if(host_set->hosts().begin(), host_set->hosts().end(),

386

66

                 std::back_inserter(*hosts_copy),

387

88

                 [&host_to_exclude](const HostSharedPtr& host) { return host_to_exclude != host; });

388

389

    // Setup a hosts to remove vector in case we need to exclude a host.

390

66

    HostVector hosts_to_remove;

391

66

    if (hosts_copy->size() != host_set->hosts().size()) {

392

5

      ASSERT(hosts_copy->size() == host_set->hosts().size() - 1);

393

5

      hosts_to_remove.emplace_back(host_to_exclude);

394

5

395

396

    // Filter hosts per locality in case we need to exclude a host.

397

66

    HostsPerLocalityConstSharedPtr hosts_per_locality_copy = host_set->hostsPerLocality().filter(

398

88

        {[&host_to_exclude](const Host& host) { return &host != host_to_exclude.get(); }})[0];

399

400

66

    prioritySet().updateHosts(

401

66

        priority, HostSetImpl::partitionHosts(hosts_copy, hosts_per_locality_copy),

402

66

        host_set->localityWeights(), {}, hosts_to_remove, absl::nullopt, absl::nullopt);

403

66

404

66

405

406

bool EdsClusterImpl::updateHostsPerLocality(

407

    const uint32_t priority, bool weighted_priority_health, const uint32_t overprovisioning_factor,

408

    const HostVector& new_hosts, LocalityWeightsMap& locality_weights_map,

409

    LocalityWeightsMap& new_locality_weights_map, PriorityStateManager& priority_state_manager,

410

944

    const HostMap& all_hosts, const absl::flat_hash_set<std::string>& all_new_hosts) {

411

944

  const auto& host_set = priority_set_.getOrCreateHostSet(priority, overprovisioning_factor);

412

944

  HostVectorSharedPtr current_hosts_copy(new HostVector(host_set.hosts()));

413

414

944

  HostVector hosts_added;

415

944

  HostVector hosts_removed;

416

944

  hosts_added.reserve(new_hosts.size());

417

944

  hosts_removed.reserve(host_set.hosts().size());

418

  // We need to trigger updateHosts with the new host vectors if they have changed. We also do this

419

  // when the locality weight map or the overprovisioning factor. Note calling updateDynamicHostList

420

  // is responsible for both determining whether there was a change and to perform the actual update

421

  // to current_hosts_copy, so it must be called even if we know that we need to update (e.g. if the

422

  // overprovisioning factor changes).

423

//

424

  // TODO(htuch): We eagerly update all the host sets here on weight changes, which may have

425

  // performance implications, since this has the knock on effect that we rebuild the load balancers

426

  // and locality scheduler. See the comment in BaseDynamicClusterImpl::updateDynamicHostList

427

  // about this. In the future we may need to do better here.

428

944

  const bool hosts_updated = updateDynamicHostList(new_hosts, *current_hosts_copy, hosts_added,

429

944

                                                   hosts_removed, all_hosts, all_new_hosts);

430

944

  if (hosts_updated || host_set.weightedPriorityHealth() != weighted_priority_health ||

431

944

      host_set.overprovisioningFactor() != overprovisioning_factor ||

432

944

      locality_weights_map != new_locality_weights_map) {

433

792

    ASSERT(std::all_of(current_hosts_copy->begin(), current_hosts_copy->end(),

434

792

                       [&](const auto& host) { return host->priority() == priority; }));

435

792

    locality_weights_map = new_locality_weights_map;

436

792

    ENVOY_LOG(debug,

437

792

              "EDS hosts or locality weights changed for cluster: {} current hosts {} priority {}",

438

792

              info_->name(), host_set.hosts().size(), host_set.priority());

439

440

792

    priority_state_manager.updateClusterPrioritySet(

441

792

        priority, std::move(current_hosts_copy), hosts_added, hosts_removed, absl::nullopt,

442

792

        weighted_priority_health, overprovisioning_factor);

443

792

    return true;

444

792

445

152

  return false;

446

944

447

448

void EdsClusterImpl::onConfigUpdateFailed(Envoy::Config::ConfigUpdateFailureReason reason,

449

31

                                          const EnvoyException*) {

450

31

  ASSERT(Envoy::Config::ConfigUpdateFailureReason::ConnectionFailure != reason);

451

  // Config failure may happen if Envoy times out waiting for the EDS resource.

452

  // If it is a timeout, the eds resources cache is enabled,

453

  // and there is a cached ClusterLoadAssignment, then the cached assignment should be used.

454

31

  if (reason == Envoy::Config::ConfigUpdateFailureReason::FetchTimedout &&

455

31

      eds_resources_cache_.has_value()) {

456

19

    ENVOY_LOG(trace, "onConfigUpdateFailed due to timeout for {}, looking for cached resources",

457

19

              edsServiceName());

458

19

    auto cached_resource = eds_resources_cache_->getResource(edsServiceName(), this);

459

19

    if (cached_resource.has_value()) {

460

19

      ENVOY_LOG(

461

19

          debug,

462

19

          "Did not receive EDS response on time, using cached ClusterLoadAssignment for cluster {}",

463

19

          edsServiceName());

464

19

      envoy::config::endpoint::v3::ClusterLoadAssignment cached_load_assignment =

465

19

          dynamic_cast<const envoy::config::endpoint::v3::ClusterLoadAssignment&>(*cached_resource);

466

19

      info_->configUpdateStats().assignment_use_cached_.inc();

467

19

      using_cached_resource_ = true;

468

19

      update(std::move(cached_load_assignment));

469

19

      return;

470

19

471

19

472

  // We need to allow server startup to continue, even if we have a bad config.

473

12

  onPreInitComplete();

474

12

475

476

absl::StatusOr<std::pair<ClusterImplBaseSharedPtr, ThreadAwareLoadBalancerPtr>>

477

EdsClusterFactory::createClusterImpl(const envoy::config::cluster::v3::Cluster& cluster,

478

531

                                     ClusterFactoryContext& context) {

479

  // TODO(kbaichoo): EDS cluster should be able to support loading it's

480

  // configuration from the CustomClusterType protobuf. Currently it does not.

481

  // See: https://github.com/envoyproxy/envoy/issues/28752

482

531

  if (!cluster.has_eds_cluster_config()) {

483

2

    return absl::InvalidArgumentError("cannot create an EDS cluster without an EDS config");

484

2

485

486

529

  absl::StatusOr<std::unique_ptr<EdsClusterImpl>> cluster_or_error =

487

529

      EdsClusterImpl::create(cluster, context);

488

529

  RETURN_IF_NOT_OK(cluster_or_error.status());

489

529

  return std::make_pair(std::move(*cluster_or_error), nullptr);

490

529

491

492

840

bool EdsClusterImpl::validateAllLedsUpdated() const {

493

  // Iterate through all LEDS based localities, and if they are all updated return true.

494

858

  for (const auto& [_, leds_subscription] : leds_localities_) {

495

119

    if (!leds_subscription->isUpdated()) {

496

40

      return false;

497

40

498

119

499

800

  return true;

500

840

501

502

/**

503

 * Static registration for the Eds cluster factory. @see RegisterFactory.

504

*/

505

REGISTER_FACTORY(EdsClusterFactory, ClusterFactory);

506

507

} // namespace Upstream

508

} // namespace Envoy