| 8044 | |
| 8045 | |
| 8046 | void Master::updateUnavailability( |
| 8047 | const MachineID& machineId, |
| 8048 | const Option<Unavailability>& unavailability) |
| 8049 | { |
| 8050 | if (unavailability.isSome()) { |
| 8051 | machines[machineId].info.mutable_unavailability()->CopyFrom( |
| 8052 | unavailability.get()); |
| 8053 | } else { |
| 8054 | machines[machineId].info.clear_unavailability(); |
| 8055 | } |
| 8056 | |
| 8057 | // TODO(jmlvanre): Only update allocator and rescind offers if the |
| 8058 | // unavailability has actually changed. |
| 8059 | if (machines.contains(machineId)) { |
| 8060 | // For every slave on this machine, update the allocator. |
| 8061 | foreach (const SlaveID& slaveId, machines[machineId].slaves) { |
| 8062 | // The slave should not be in the machines mapping if it is removed. |
| 8063 | CHECK(slaves.removed.get(slaveId).isNone()); |
| 8064 | |
| 8065 | // The slave should be registered if it is in the machines mapping. |
| 8066 | CHECK(slaves.registered.contains(slaveId)); |
| 8067 | |
| 8068 | Slave* slave = slaves.registered.get(slaveId); |
| 8069 | |
| 8070 | if (unavailability.isSome()) { |
| 8071 | // TODO(jmlvanre): Add stream operator for unavailability. |
| 8072 | LOG(INFO) << "Updating unavailability of agent " << *slave |
| 8073 | << ", starting at " |
| 8074 | << Nanoseconds(unavailability->start().nanoseconds()); |
| 8075 | } else { |
| 8076 | LOG(INFO) << "Removing unavailability of agent " << *slave; |
| 8077 | } |
| 8078 | |
| 8079 | // Rescind offers since we want to inform frameworks of the |
| 8080 | // unavailability change as soon as possible. |
| 8081 | foreach (Offer* offer, utils::copy(slave->offers)) { |
| 8082 | rescindOffer(offer); |
| 8083 | } |
| 8084 | |
| 8085 | // Remove and rescind inverse offers since the allocator will send new |
| 8086 | // inverse offers for the updated unavailability. |
| 8087 | foreach (InverseOffer* inverseOffer, utils::copy(slave->inverseOffers)) { |
| 8088 | allocator->updateInverseOffer( |
| 8089 | slave->id, |
| 8090 | inverseOffer->framework_id(), |
| 8091 | UnavailableResources{ |
| 8092 | inverseOffer->resources(), |
| 8093 | inverseOffer->unavailability()}, |
| 8094 | None()); |
| 8095 | |
| 8096 | removeInverseOffer(inverseOffer, true); // Rescind! |
| 8097 | } |
| 8098 | |
| 8099 | // We remove / rescind all the offers first so that any calls to the |
| 8100 | // allocator to modify its internal state are queued before the update of |
| 8101 | // the unavailability in the allocator. We do this so that the allocator's |
| 8102 | // state can start from a "clean slate" for the new unavailability. |
| 8103 | // NOTE: Any calls from the Allocator back into the master, for example |