| 2314 | } |
| 2315 | |
| 2316 | void AdmissionController::UpdateClusterAggregates(const set<string>& removed_nodes) { |
| 2317 | // Recompute mem_reserved for all hosts. |
| 2318 | PoolStats::HostMemMap updated_mem_reserved; |
| 2319 | for (auto& entry : pool_stats_) { |
| 2320 | entry.second.UpdateAggregates(&updated_mem_reserved); |
| 2321 | } |
| 2322 | |
| 2323 | root_agg_user_loads_.clear(); |
| 2324 | for (auto& entry : pool_stats_) { |
| 2325 | root_agg_user_loads_.add_loads( |
| 2326 | entry.second.get_aggregated_user_loads().get_user_loads()); |
| 2327 | } |
| 2328 | |
| 2329 | stringstream ss; |
| 2330 | ss << "Updated mem reserved for hosts:"; |
| 2331 | int i = 0; |
| 2332 | for (const auto& e : updated_mem_reserved) { |
| 2333 | int64_t old_mem_reserved = host_stats_[e.first].mem_reserved; |
| 2334 | if (old_mem_reserved == e.second) continue; |
| 2335 | host_stats_[e.first].mem_reserved = e.second; |
| 2336 | if (VLOG_ROW_IS_ON) { |
| 2337 | ss << endl << e.first << ": " << PrintBytes(old_mem_reserved); |
| 2338 | ss << " -> " << PrintBytes(e.second); |
| 2339 | ++i; |
| 2340 | } |
| 2341 | } |
| 2342 | |
| 2343 | // We know if any host stats were removed from the pool stats during statestore topic |
| 2344 | // update by the set pool_stats_removed_nodes. If a host was removed and no longer |
| 2345 | // exists in any remote pool stats, we reset the mem_reserved to 0 for this host in |
| 2346 | // the host stats. |
| 2347 | for (const auto& host : removed_nodes) { |
| 2348 | auto it = host_stats_.find(host); |
| 2349 | if (it != host_stats_.end() |
| 2350 | && updated_mem_reserved.find(host) == updated_mem_reserved.end()) { |
| 2351 | int64_t old_mem_reserved = it->second.mem_reserved; |
| 2352 | it->second.mem_reserved = 0; |
| 2353 | if (VLOG_ROW_IS_ON) { |
| 2354 | ss << endl |
| 2355 | << "Mem_reserved reset to 0 for removed host: " << host |
| 2356 | << " (old value=" << PrintBytes(old_mem_reserved) << ")"; |
| 2357 | ++i; |
| 2358 | } |
| 2359 | } |
| 2360 | } |
| 2361 | |
| 2362 | if (i > 0) VLOG_ROW << ss.str(); |
| 2363 | } |
| 2364 | |
| 2365 | Status AdmissionController::ComputeGroupScheduleStates( |
| 2366 | const ClusterMembershipMgr::SnapshotPtr& membership_snapshot, QueueNode* queue_node) { |