upstream: fix bug causing hosts to be dropped (envoyproxy#4476)

Previously when a host was moved from an existing priority to a new
priority, the update would not cause a host update, causing the host to
essentially disappear. This was happening because the code would
determine that it knew about the host already (it existed in another
priority), but due to a lack of distinction between whether the
existing host came from the same priority (should not trigger an update)
or another (should trigger another) it would not update the host list when
the only new host came from another priority, leaving it empty.

The fix involves comparing the list of hosts for the current
priority with the list of hosts matched to an existing host. By being in
the list of existing hosts, the hosts must either exist in the current
priority or another. By comparing these lists we can determine whether
the matched host came from a different priority, in which case we mark
the host list as changed.

Signed-off-by: Snow Pettersen [email protected]

Risk Level: Medium, changes host update behavior
Testing: Added regression test
Docs Changes: n/a
Release Notes: n/a
Fixes envoyproxy#4473

source/common/upstream/upstream_impl.cc

@@ -1004,13 +1004,22 @@ bool BaseDynamicClusterImpl::updateDynamicHostList(
 
   // Remove hosts from current_priority_hosts that were matched to an existing host in the previous
   // loop.
-  current_priority_hosts.erase(std::remove_if(current_priority_hosts.begin(),
-                                              current_priority_hosts.end(),
-                                              [&existing_hosts_for_current_priority](auto host) {
-                                                return existing_hosts_for_current_priority.count(
-                                                    host->address()->asString());
-                                              }),
-                               current_priority_hosts.end());
+  for (auto itr = current_priority_hosts.begin(); itr != current_priority_hosts.end();) {
+    auto existing_itr = existing_hosts_for_current_priority.find((*itr)->address()->asString());
+
+    if (existing_itr != existing_hosts_for_current_priority.end()) {
+      existing_hosts_for_current_priority.erase(existing_itr);
+      itr = current_priority_hosts.erase(itr);
+    } else {
+      itr++;
+    }
+  }
+
+  // If we saw existing hosts during this iteration from a different priority, then we've moved
+  // a host from another priority into this one, so we should mark the priority as having changed.
+  if (!existing_hosts_for_current_priority.empty()) {
+    hosts_changed = true;
+  }
 
   // The remaining hosts are hosts that are not referenced in the config update. We remove them from
   // the priority if any of the following is true:

test/common/upstream/eds_test.cc

@@ -481,6 +481,92 @@ TEST_F(EdsTest, EndpointRemoval) {
   }
 }
 
+// Verifies that if an endpoint is moved to a new priority, the active hc status is preserved.
+TEST_F(EdsTest, EndpointMovedToNewPriority) {
+  resetCluster(R"EOF(
+      name: name
+      connect_timeout: 0.25s
+      type: EDS
+      lb_policy: ROUND_ROBIN
+      drain_connections_on_host_removal: true
+      eds_cluster_config:
+        service_name: fare
+        eds_config:
+          api_config_source:
+            cluster_names:
+            - eds
+            refresh_delay: 1s
+  )EOF");
+
+  auto health_checker = std::make_shared<MockHealthChecker>();
+  EXPECT_CALL(*health_checker, start());
+  EXPECT_CALL(*health_checker, addHostCheckCompleteCb(_)).Times(2);
+  cluster_->setHealthChecker(health_checker);
+
+  Protobuf::RepeatedPtrField<envoy::api::v2::ClusterLoadAssignment> resources;
+  auto* cluster_load_assignment = resources.Add();
+  cluster_load_assignment->set_cluster_name("fare");
+
+  auto add_endpoint = [cluster_load_assignment](int port, int priority) {
+    auto* endpoints = cluster_load_assignment->add_endpoints();
+    endpoints->set_priority(priority);
+
+    auto* socket_address = endpoints->add_lb_endpoints()
+                               ->mutable_endpoint()
+                               ->mutable_address()
+                               ->mutable_socket_address();
+    socket_address->set_address("1.2.3.4");
+    socket_address->set_port_value(port);
+  };
+
+  add_endpoint(80, 0);
+  add_endpoint(81, 0);
+
+  VERBOSE_EXPECT_NO_THROW(cluster_->onConfigUpdate(resources, ""));
+
+  {
+    auto& hosts = cluster_->prioritySet().hostSetsPerPriority()[0]->hosts();
+    EXPECT_EQ(hosts.size(), 2);
+
+    // Mark the hosts as healthy
+    for (auto& host : hosts) {
+      EXPECT_TRUE(host->healthFlagGet(Host::HealthFlag::FAILED_ACTIVE_HC));
+      host->healthFlagClear(Host::HealthFlag::FAILED_ACTIVE_HC);
+    }
+  }
+
+  // Moves the endpoints between priorities
+  cluster_load_assignment->clear_endpoints();
+  add_endpoint(81, 0);
+  add_endpoint(80, 1);
+
+  VERBOSE_EXPECT_NO_THROW(cluster_->onConfigUpdate(resources, ""));
+
+  {
+    auto& hosts = cluster_->prioritySet().hostSetsPerPriority()[0]->hosts();
+    EXPECT_EQ(hosts.size(), 1);
+
+    // assert that it didn't move
+    EXPECT_EQ(hosts[0]->address()->asString(), "1.2.3.4:81");
+
+    // The endpoint was healthy in the original priority, so moving it
+    // around should preserve that.
+    EXPECT_FALSE(hosts[0]->healthFlagGet(Host::HealthFlag::FAILED_ACTIVE_HC));
+  }
+
+  {
+    auto& hosts = cluster_->prioritySet().hostSetsPerPriority()[1]->hosts();
+    EXPECT_EQ(hosts.size(), 1);
+
+    // assert that it moved
+    EXPECT_EQ(hosts[0]->address()->asString(), "1.2.3.4:80");
+
+    // The endpoint was healthy in the original priority, so moving it
+    // around should preserve that.
+    EXPECT_FALSE(hosts[0]->healthFlagGet(Host::HealthFlag::FAILED_ACTIVE_HC));
+  }
+}
+
 // Verifies that if an endpoint is moved between priorities, the health check value
 // of the host is preserved
 TEST_F(EdsTest, EndpointMoved) {