Added new `--reconfiguration_policy` slave flag and implementation. This flag allows operators to select a set of permitted configuration changes that the slave will tolerate during recovery while still recovering running tasks and keeping the same agent id.
The previous behaviour of Mesos is reproduced exactly by setting this flag to "equal". For now only one additional policy is provided, "additive". Review: https://reviews.apache.org/r/64012/ Project: http://git-wip-us.apache.org/repos/asf/mesos/repo Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/63e3336b Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/63e3336b Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/63e3336b Branch: refs/heads/master Commit: 63e3336b508f5fe3936ed701404cde7d70eee754 Parents: f4d2c8f Author: Benno Evers <bev...@mesosphere.com> Authored: Tue Dec 5 13:56:00 2017 -0800 Committer: Vinod Kone <vinodk...@gmail.com> Committed: Tue Dec 5 13:58:39 2017 -0800 ---------------------------------------------------------------------- src/CMakeLists.txt | 1 + src/Makefile.am | 2 + src/slave/compatibility.cpp | 207 +++++++++++++++++++++ src/slave/compatibility.hpp | 66 +++++++ src/slave/flags.cpp | 14 ++ src/slave/flags.hpp | 1 + src/slave/slave.cpp | 119 +++++++++---- src/slave/slave.hpp | 9 + src/tests/CMakeLists.txt | 1 + src/tests/master_tests.cpp | 257 ++++++++++++++++++++++++++- src/tests/slave_compatibility_tests.cpp | 175 ++++++++++++++++++ src/tests/slave_recovery_tests.cpp | 135 ++++++++++++++ src/tests/slave_tests.cpp | 98 ++++++++++ 13 files changed, 1052 insertions(+), 33 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 592489d..a76ba1e 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -246,6 +246,7 @@ configure_file( # SOURCE FILES FOR THE MESOS LIBRARY. ##################################### set(AGENT_SRC + slave/compatibility.cpp slave/constants.cpp slave/container_daemon.cpp slave/container_logger.cpp http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/Makefile.am ---------------------------------------------------------------------- diff --git a/src/Makefile.am b/src/Makefile.am index d5ca797..05e8b95 100644 --- a/src/Makefile.am +++ b/src/Makefile.am @@ -1015,6 +1015,7 @@ libmesos_no_3rdparty_la_SOURCES += \ sched/sched.cpp \ scheduler/scheduler.cpp \ secret/resolver.cpp \ + slave/compatibility.cpp \ slave/constants.cpp \ slave/container_daemon.cpp \ slave/container_logger.cpp \ @@ -2494,6 +2495,7 @@ mesos_tests_SOURCES = \ tests/scheduler_tests.cpp \ tests/script.cpp \ tests/slave_authorization_tests.cpp \ + tests/slave_compatibility_tests.cpp \ tests/slave_recovery_tests.cpp \ tests/slave_validation_tests.cpp \ tests/slave_tests.cpp \ http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/slave/compatibility.cpp ---------------------------------------------------------------------- diff --git a/src/slave/compatibility.cpp b/src/slave/compatibility.cpp new file mode 100644 index 0000000..4ead4a5 --- /dev/null +++ b/src/slave/compatibility.cpp @@ -0,0 +1,207 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "slave/compatibility.hpp" + +#include <stout/strings.hpp> +#include <stout/unreachable.hpp> + +#include <mesos/values.hpp> + +#include "mesos/type_utils.hpp" + +namespace mesos { +namespace internal { +namespace slave { +namespace compatibility { + +// TODO(bevers): Compare the SlaveInfo fields individually, in order to be +// able to generate better error messages. +Try<Nothing> equal( + const SlaveInfo& previous, + const SlaveInfo& current) +{ + if (previous == current) { + return Nothing(); + } + + return Error(strings::join( + "\n", + "Incompatible agent info detected. ", + "\n------------------------------------------------------------", + "Old agent info:\n" + stringify(previous), + "\n------------------------------------------------------------", + "New agent info:\n" + stringify(current), + "\n------------------------------------------------------------")); +} + + +// T is instantiated below as either `Resource` or `Attribute`. +template<typename T> +Try<T> getMatchingValue( + const T& previous, + const google::protobuf::RepeatedPtrField<T>& values) +{ + auto match = std::find_if( + values.begin(), + values.end(), + [&previous](const T& value) { + return previous.name() == value.name(); + }); + + if (match == values.end()) { + return Error("Couldn't find '" + previous.name() + "'"); + } + + if (match->type() != previous.type()) { + return Error( + "Type of '" + previous.name() + "' changed from " + + stringify(previous.type()) + " to " + stringify(match->type())); + } + + return *match; +} + + +Try<Nothing> additive( + const SlaveInfo& previous, + const SlaveInfo& current) +{ + if (previous.hostname() != current.hostname()) { + return Error( + "Configuration change not permitted under `additive` policy: " + "Hostname changed from " + + previous.hostname() + " to " + current.hostname()); + } + + if (previous.port() != current.port()) { + return Error( + "Configuration change not permitted under `additive` policy: " + "Port changed from " + stringify(previous.port()) + " to " + + stringify(current.port())); + } + + if (previous.has_domain() && !(previous.domain() == current.domain())) { + return Error( + "Configuration change not permitted under `additive` policy: " + "Domain changed from " + stringify(previous.domain()) + " to " + + stringify(current.domain())); + } + + // TODO(bennoe): We should probably check `resources.size()` and switch to a + // smarter algorithm for the matching when its bigger than, say, 20. + for (const Resource& resource : previous.resources()) { + Try<Resource> match = + getMatchingValue(resource, current.resources()); + + if (match.isError()) { + return Error( + "Configuration change not permitted under 'additive' policy: " + + match.error()); + } + + switch (resource.type()) { + case Value::SCALAR: { + if (!(resource.scalar() <= match->scalar())) { + return Error( + "Configuration change not permitted under 'additive' policy: " + "Value of scalar resource '" + resource.name() + "' decreased " + "from " + stringify(resource.scalar()) + " to " + + stringify(match->scalar())); + } + continue; + } + case Value::RANGES: { + if (!(resource.ranges() <= match->ranges())) { + return Error( + "Configuration change not permitted under 'additive' policy: " + "Previous value of range resource '" + resource.name() + "' (" + + stringify(resource.ranges()) + ") not included in current " + + stringify(match->ranges())); + } + continue; + } + case Value::SET: { + if (!(resource.set() <= match->set())) { + return Error( + "Configuration change not permitted under 'additive' policy: " + "Previous value of set resource '" + resource.name() + "' (" + + stringify(resource.set()) + ") not included in current " + + stringify(match->set())); + } + continue; + } + case Value::TEXT: { + // Text resources are not supported. + UNREACHABLE(); + } + } + } + + for (const Attribute& attribute : previous.attributes()) { + Try<Attribute> match = + getMatchingValue(attribute, current.attributes()); + + if (match.isError()) { + return Error( + "Configuration change not permitted under 'additive' policy: " + + match.error()); + } + + switch (attribute.type()) { + case Value::SCALAR: { + if (!(attribute.scalar() == match->scalar())) { + return Error( + "Configuration change not permitted under 'additive' policy: " + "Value of scalar attribute '" + attribute.name() + "' changed " + "from " + stringify(attribute.scalar()) + " to " + + stringify(match->scalar())); + } + continue; + } + case Value::RANGES: { + if (!(attribute.ranges() <= match->ranges())) { + return Error( + "Previous value of ranges resource '" + attribute.name() + "' (" + + stringify(attribute.ranges()) + ") not included in current " + + stringify(match->ranges())); + } + continue; + } + case Value::TEXT: { + if (!(attribute.text() == match->text())) { + return Error( + "Configuration change not permitted under 'additive' policy: " + "Value of text attribute '" + attribute.name() + "' changed " + "from " + stringify(attribute.text()) + + " to " + stringify(match->text())); + } + continue; + } + case Value::SET: { + // Set attributes are not supported. + UNREACHABLE(); + } + } + } + + return Nothing(); +} + +} // namespace compatibility { +} // namespace slave { +} // namespace internal { +} // namespace mesos { http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/slave/compatibility.hpp ---------------------------------------------------------------------- diff --git a/src/slave/compatibility.hpp b/src/slave/compatibility.hpp new file mode 100644 index 0000000..78b421a --- /dev/null +++ b/src/slave/compatibility.hpp @@ -0,0 +1,66 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef __SLAVE_COMPATIBILITY_HPP__ +#define __SLAVE_COMPATIBILITY_HPP__ + +#include <stout/nothing.hpp> +#include <stout/try.hpp> + +#include <mesos/mesos.pb.h> + +namespace mesos { +namespace internal { +namespace slave { +namespace compatibility { + +// This function checks whether `previous` and `current` are considered to be +// "equal", i.e., all fields in `SlaveInfo` are the same. +Try<Nothing> equal( + const SlaveInfo& previous, + const SlaveInfo& current); + + +// This function checks whether the changes between `previous` and `current` +// are considered to be "additive", according to the rules in the following +// table: +// +// Field | Constraint +// ----------------------------------------------------------------------------- +// hostname | Must match exactly. +// port | Must match exactly. +// domain | Must either match exactly or change from not configured to +// | configured. +// resources | All previous resources must be present with the same type. +// | For type SCALAR: The new value must be not smaller than the old. +// | For type RANGE: The new value must include the old ranges. +// | For type SET: The new value must be a superset of the old. +// | New resources are permitted. +// attributes | All previous attributes must be present with the same type. +// | For type SCALAR: The new value must be not smaller than the old. +// | For type RANGE: The new value must include the old ranges. +// | For type TEXT: The new value must exactly match the previous. +// | New attributes are permitted. +Try<Nothing> additive( + const SlaveInfo& previous, + const SlaveInfo& current); + +} // namespace compatibility { +} // namespace slave { +} // namespace internal { +} // namespace mesos { + +#endif // __SLAVE_COMPATIBILITY_HPP__ http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/slave/flags.cpp ---------------------------------------------------------------------- diff --git a/src/slave/flags.cpp b/src/slave/flags.cpp index 34edf4c..d876474 100644 --- a/src/slave/flags.cpp +++ b/src/slave/flags.cpp @@ -459,6 +459,20 @@ mesos::internal::slave::Flags::Flags() "sucessfully reconnect to the framework.", RECOVERY_TIMEOUT); + add(&Flags::reconfiguration_policy, + "reconfiguration_policy", + "This flag controls which agent configuration changes are considered\n" + "acceptable when recovering the previous agent state. Possible values:\n" + "equal: Require that the old and the new state match exactly.\n" + "additive: Require that the new state is a superset of the old state:\n" + " it is permitted to add additional resources, attributes\n" + " and domains but not to remove existing ones.\n" + "Note that this only affects the checking done on the agent itself,\n" + "the master may still reject the slave if it detects a change that it\n" + "considers unacceptable, which currently happens when port or hostname\n" + "are changed.", + "equal"); + add(&Flags::strict, "strict", "If `strict=true`, any and all recovery errors are considered fatal.\n" http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/slave/flags.hpp ---------------------------------------------------------------------- diff --git a/src/slave/flags.hpp b/src/slave/flags.hpp index f25d8af..f84ba5a 100644 --- a/src/slave/flags.hpp +++ b/src/slave/flags.hpp @@ -90,6 +90,7 @@ public: Option<std::string> container_logger; + std::string reconfiguration_policy; std::string recover; Duration recovery_timeout; bool strict; http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/slave/slave.cpp ---------------------------------------------------------------------- diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp index c07e25f..4927001 100644 --- a/src/slave/slave.cpp +++ b/src/slave/slave.cpp @@ -90,8 +90,11 @@ #include "logging/logging.hpp" +#include "master/detector/standalone.hpp" + #include "module/manager.hpp" +#include "slave/compatibility.hpp" #include "slave/constants.hpp" #include "slave/flags.hpp" #include "slave/paths.hpp" @@ -842,6 +845,15 @@ void Slave::initialize() } } + // Check that the reconfiguration_policy flag is valid. + if (flags.reconfiguration_policy != "equal" && + flags.reconfiguration_policy != "additive") { + EXIT(EXIT_FAILURE) + << "Unknown option for 'reconfiguration_policy' flag " + << flags.reconfiguration_policy << "." + << " Please run the agent with '--help' to see the valid options."; + } + // Check that the recover flag is valid. if (flags.recover != "reconnect" && flags.recover != "cleanup") { EXIT(EXIT_FAILURE) @@ -1010,6 +1022,26 @@ void Slave::detected(const Future<Option<MasterInfo>>& _master) return; } + if (requiredMasterCapabilities.agentUpdate) { + protobuf::master::Capabilities masterCapabilities( + latest->capabilities()); + + if (!masterCapabilities.agentUpdate) { + EXIT(EXIT_FAILURE) << + "Agent state changed on restart, but the detected master lacks the " + "AGENT_UPDATE capability. Refusing to connect."; + return; + } + + if (dynamic_cast<mesos::master::detector::StandaloneMasterDetector*>( + detector)) { + LOG(WARNING) << + "The AGENT_UPDATE master capability is required, " + "but the StandaloneMasterDetector does not have the ability to read " + "master capabilities."; + } + } + // Wait for a random amount of time before authentication or // registration. Duration duration = @@ -6143,6 +6175,25 @@ void Slave::_checkDiskUsage(const Future<double>& usage) } +Try<Nothing> Slave::compatible( + const SlaveInfo& previous, + const SlaveInfo& current) const +{ + // TODO(vinod): Also check for version compatibility. + + if (flags.reconfiguration_policy == "equal") { + return compatibility::equal(previous, current); + } + + if (flags.reconfiguration_policy == "additive") { + return compatibility::additive(previous, current); + } + + // Should have been validated during startup. + UNREACHABLE(); +} + + Future<Nothing> Slave::recover(const Try<state::State>& state) { if (state.isError()) { @@ -6303,43 +6354,53 @@ Future<Nothing> Slave::recover(const Try<state::State>& state) metrics.recovery_errors += slaveState->errors; } + // Save the previous id into the current `SlaveInfo`, so we can compare + // both of them for equality. This is safe because if it turned out that + // we can not reuse the id, we will either crash or erase it again. + info.mutable_id()->CopyFrom(slaveState->info->id()); + // Check for SlaveInfo compatibility. - // TODO(vinod): Also check for version compatibility. - - SlaveInfo _info(info); - _info.mutable_id()->CopyFrom(slaveState->id); - if (flags.recover == "reconnect" && - !(_info == slaveState->info.get())) { - string message = strings::join( - "\n", - "Incompatible agent info detected.", - "------------------------------------------------------------", - "Old agent info:\n" + stringify(slaveState->info.get()), - "------------------------------------------------------------", - "New agent info:\n" + stringify(info), - "------------------------------------------------------------"); - - // Fail the recovery unless the agent is recovering for the first - // time after host reboot. - // + Try<Nothing> _compatible = + compatible(slaveState->info.get(), info); + + if (_compatible.isSome()) { + // Permitted change, so we reuse the recovered agent id and reconnect + // to running executors. + + // Prior to Mesos 1.5, the master expected that an agent would never + // change its `SlaveInfo` and keep the same slave id, and therefore would + // not update it's internal data structures on agent re-registration. + if (!(slaveState->info.get() == info)) { + requiredMasterCapabilities.agentUpdate = true; + } + + // Start the local resource providers daemon once we have the slave id. + localResourceProviderDaemon->start(info.id()); + + // Recover the frameworks. + foreachvalue (const FrameworkState& frameworkState, + slaveState->frameworks) { + recoverFramework(frameworkState, injectedExecutors, injectedTasks); + } + } else if (state->rebooted) { // Prior to Mesos 1.4 we directly bypass the state recovery and // start as a new agent upon reboot (introduced in MESOS-844). // This unncessarily discards the existing agent ID (MESOS-6223). // Starting in Mesos 1.4 we'll attempt to recover the slave state - // even after reboot but in case of slave info mismatch we'll fall - // back to recovering as a new agent (existing behavior). This - // prevents the agent from flapping if the slave info (resources, + // even after reboot but in case of an incompatible slave info change + // we'll fall back to recovering as a new agent (existing behavior). + // Prior to Mesos 1.5, an incompatible change would be any slave info + // mismatch. + // This prevents the agent from flapping if the slave info (resources, // attributes, etc.) change is due to host maintenance associated // with the reboot. - if (!state->rebooted) { - return Failure(message); - } LOG(WARNING) << "Falling back to recover as a new agent due to error: " - << message; + << _compatible.error(); // Cleaning up the slave state to avoid any state recovery for the // old agent. + info.clear_id(); slaveState = None(); // Remove the "latest" symlink if it exists to "checkpoint" the @@ -6350,13 +6411,7 @@ Future<Nothing> Slave::recover(const Try<state::State>& state) << "Failed to remove latest symlink '" << latest << "'"; } } else { - info = slaveState->info.get(); // Recover the slave info. - - // Recover the frameworks. - foreachvalue (const FrameworkState& frameworkState, - slaveState->frameworks) { - recoverFramework(frameworkState, injectedExecutors, injectedTasks); - } + return Failure(_compatible.error()); } } http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/slave/slave.hpp ---------------------------------------------------------------------- diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp index 643d855..06afd52 100644 --- a/src/slave/slave.hpp +++ b/src/slave/slave.hpp @@ -588,6 +588,15 @@ private: double _resources_revocable_used(const std::string& name); double _resources_revocable_percent(const std::string& name); + // Checks whether the two `SlaveInfo` objects are considered + // compatible based on the value of the `--configuration_policy` + // flag. + Try<Nothing> compatible( + const SlaveInfo& previous, + const SlaveInfo& current) const; + + protobuf::master::Capabilities requiredMasterCapabilities; + const Flags flags; const Http http; http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/tests/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/src/tests/CMakeLists.txt b/src/tests/CMakeLists.txt index 5e52020..b74fbb9 100644 --- a/src/tests/CMakeLists.txt +++ b/src/tests/CMakeLists.txt @@ -115,6 +115,7 @@ set(MESOS_TESTS_SRC scheduler_http_api_tests.cpp scheduler_tests.cpp slave_authorization_tests.cpp + slave_compatibility_tests.cpp slave_tests.cpp slave_validation_tests.cpp sorter_tests.cpp http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/tests/master_tests.cpp ---------------------------------------------------------------------- diff --git a/src/tests/master_tests.cpp b/src/tests/master_tests.cpp index 64ac2d5..0dfa8c7 100644 --- a/src/tests/master_tests.cpp +++ b/src/tests/master_tests.cpp @@ -2655,6 +2655,261 @@ TEST_F(MasterTest, SlavesEndpointQuerySlave) } +// Tests that the master correctly updates the slave info on +// slave re-registration. +TEST_F(MasterTest, RegistryUpdateAfterReconfiguration) +{ + // Start a master. + master::Flags masterFlags = CreateMasterFlags(); + masterFlags.registry = "replicated_log"; + Try<Owned<cluster::Master>> master = StartMaster(masterFlags); + ASSERT_SOME(master); + + // Start a slave. + Future<SlaveRegisteredMessage> slaveRegisteredMessage = + FUTURE_PROTOBUF(SlaveRegisteredMessage(), master.get()->pid, _); + + // Reuse slaveFlags so both StartSlave() use the same work_dir. + slave::Flags slaveFlags = this->CreateSlaveFlags(); + slaveFlags.resources = "cpus:100"; + + Owned<MasterDetector> detector = master.get()->createDetector(); + Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags); + ASSERT_SOME(slave); + + AWAIT_READY(slaveRegisteredMessage); + + // Restart slave with changed resources. + slave.get()->terminate(); + slave->reset(); + + Future<SlaveReregisteredMessage> slaveReregisteredMessage = + FUTURE_PROTOBUF(SlaveReregisteredMessage(), master.get()->pid, _); + + slaveFlags.reconfiguration_policy = "additive"; + slaveFlags.resources = "cpus:200"; + slave = StartSlave(detector.get(), slaveFlags); + ASSERT_SOME(slave); + + AWAIT_READY(slaveReregisteredMessage); + + // Verify master has correctly updated the slave state. + { + string slaveId = slaveReregisteredMessage->slave_id().value(); + + Future<Response> response = process::http::get( + master.get()->pid, + "slaves?slave_id=" + slaveId, + None(), + createBasicAuthHeaders(DEFAULT_CREDENTIAL)); + + AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response); + AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response); + + const Try<JSON::Value> value = JSON::parse<JSON::Value>(response->body); + + ASSERT_SOME(value); + + Try<JSON::Object> object = value->as<JSON::Object>(); + + Result<JSON::Array> array = object->find<JSON::Array>("slaves"); + ASSERT_SOME(array); + EXPECT_EQ(1u, array->values.size()); + + Try<JSON::Value> expected = JSON::parse( + "{" + "\"slaves\":" + "[{" + "\"id\":\"" + slaveId + "\"," + "\"resources\": {\"cpus\": 200}" + "}]" + "}"); + + ASSERT_SOME(expected); + + EXPECT_TRUE(value->contains(expected.get())); + } + + // Verify master has correctly updated the registry. + { + Future<Response> response = process::http::get( + master.get()->registrar.get()->pid(), + "registry", + None(), + createBasicAuthHeaders(DEFAULT_CREDENTIAL)); + + AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response); + AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response); + + const Try<JSON::Value> value = JSON::parse<JSON::Value>(response->body); + + ASSERT_SOME(value); + + Result<JSON::Array> array = value->as<JSON::Object>() + .find<JSON::Object>("slaves")->find<JSON::Array>("slaves"); + + ASSERT_SOME(array); + ASSERT_EQ(1u, array->values.size()); + + Result<JSON::Object> info = + array->values.at(0).as<JSON::Object>().find<JSON::Object>("info"); + ASSERT_SOME(info); + + string slaveId = slaveReregisteredMessage->slave_id().value(); + Result<JSON::String> id = + info->find<JSON::Object>("id")->at<JSON::String>("value"); + ASSERT_SOME(id); + ASSERT_EQ(id->value, slaveId); + + Result<JSON::Array> resources = info->find<JSON::Array>("resources"); + ASSERT_SOME(resources); + + JSON::Value expectedCpu = JSON::parse( + "{\"name\": \"cpus\"," + " \"scalar\": {\"value\": 200.0}," + " \"type\":\"SCALAR\"}").get(); + + bool found = std::any_of(resources->values.begin(), resources->values.end(), + [&](const JSON::Value& value) { return value.contains(expectedCpu); }); + + EXPECT_TRUE(found); + } +} + + +// Tests that the master correctly updates the slave state on +// slave re-registration after master failover. This is almost an exact +// duplicate of the `MasterTest.RegistryUpdateAfterReconfiguration` above, +// except that we shutdown the master prior to reconfiguring the slave. +TEST_F(MasterTest, RegistryUpdateAfterMasterFailover) +{ + // Start a master. + master::Flags masterFlags = CreateMasterFlags(); + masterFlags.registry = "replicated_log"; + Try<Owned<cluster::Master>> master = StartMaster(masterFlags); + ASSERT_SOME(master); + + // Start a slave. + Future<SlaveRegisteredMessage> slaveRegisteredMessage = + FUTURE_PROTOBUF(SlaveRegisteredMessage(), master.get()->pid, _); + + // Reuse `slaveFlags` so both `StartSlave()` calls use the same `work_dir`. + slave::Flags slaveFlags = this->CreateSlaveFlags(); + slaveFlags.resources = "cpus:100"; + + Owned<MasterDetector> detector = master.get()->createDetector(); + Try<Owned<cluster::Slave>> slave = StartSlave(detector.get(), slaveFlags); + ASSERT_SOME(slave); + + AWAIT_READY(slaveRegisteredMessage); + + // Shutdown the master. + master->reset(); + + // Shutdown the slave. + slave.get()->terminate(); + slave->reset(); + + // Restart the master. + master = StartMaster(masterFlags); + ASSERT_SOME(master); + detector = master.get()->createDetector(); + + // Restart the slave with changed resources. + Future<SlaveReregisteredMessage> slaveReregisteredMessage = + FUTURE_PROTOBUF(SlaveReregisteredMessage(), master.get()->pid, _); + + slaveFlags.reconfiguration_policy = "additive"; + slaveFlags.resources = "cpus:200"; + slave = StartSlave(detector.get(), slaveFlags); + ASSERT_SOME(slave); + + AWAIT_READY(slaveReregisteredMessage); + + // Verify master has correctly updated the slave info. + { + string slaveId = slaveReregisteredMessage->slave_id().value(); + + Future<Response> response = process::http::get( + master.get()->pid, + "slaves?slave_id=" + slaveId, + None(), + createBasicAuthHeaders(DEFAULT_CREDENTIAL)); + + AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response); + AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response); + + const Try<JSON::Value> value = JSON::parse<JSON::Value>(response->body); + + ASSERT_SOME(value); + + Try<JSON::Object> object = value->as<JSON::Object>(); + + Result<JSON::Array> array = object->find<JSON::Array>("slaves"); + ASSERT_SOME(array); + EXPECT_EQ(1u, array->values.size()); + + Try<JSON::Value> expected = JSON::parse( + "{" + "\"slaves\":" + "[{" + "\"id\":\"" + slaveId + "\"," + "\"resources\": {\"cpus\": 200}" + "}]" + "}"); + + ASSERT_SOME(expected); + + EXPECT_TRUE(value->contains(expected.get())); + } + + // Verify master has correctly updated the registry. + { + Future<Response> response = process::http::get( + master.get()->registrar.get()->pid(), + "registry", + None(), + createBasicAuthHeaders(DEFAULT_CREDENTIAL)); + + AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response); + AWAIT_EXPECT_RESPONSE_HEADER_EQ(APPLICATION_JSON, "Content-Type", response); + + const Try<JSON::Value> value = JSON::parse<JSON::Value>(response->body); + + ASSERT_SOME(value); + + Result<JSON::Array> array = value->as<JSON::Object>() + .find<JSON::Object>("slaves")->find<JSON::Array>("slaves"); + + ASSERT_SOME(array); + ASSERT_EQ(1u, array->values.size()); + + Result<JSON::Object> info = + array->values.at(0).as<JSON::Object>().find<JSON::Object>("info"); + ASSERT_SOME(info); + + string slaveId = slaveReregisteredMessage->slave_id().value(); + Result<JSON::String> id = + info->find<JSON::Object>("id")->at<JSON::String>("value"); + ASSERT_SOME(id); + ASSERT_EQ(id->value, slaveId); + + Result<JSON::Array> resources = info->find<JSON::Array>("resources"); + ASSERT_SOME(resources); + + JSON::Value expectedCpu = JSON::parse( + "{\"name\": \"cpus\"," + " \"scalar\": {\"value\": 200.0}," + " \"type\":\"SCALAR\"}").get(); + + bool found = std::any_of(resources->values.begin(), resources->values.end(), + [&](const JSON::Value& value) { return value.contains(expectedCpu); }); + + EXPECT_TRUE(found); + } +} + + // This test ensures that when a slave is recovered from the registry // but does not re-register with the master, it is marked unreachable // in the registry, the framework is informed that the slave is lost, @@ -2998,7 +3253,7 @@ TEST_F(MasterTest, CancelRecoveredSlaveRemoval) Future<SlaveRegisteredMessage> slaveRegisteredMessage = FUTURE_PROTOBUF(SlaveRegisteredMessage(), master.get()->pid, _); - // Reuse slaveFlags so both StartSlave() use the same work_dir. + // Reuse `slaveFlags` so both `StartSlave()` calls use the same `work_dir`. slave::Flags slaveFlags = CreateSlaveFlags(); Owned<MasterDetector> detector = master.get()->createDetector(); http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/tests/slave_compatibility_tests.cpp ---------------------------------------------------------------------- diff --git a/src/tests/slave_compatibility_tests.cpp b/src/tests/slave_compatibility_tests.cpp new file mode 100644 index 0000000..ab5ed29 --- /dev/null +++ b/src/tests/slave_compatibility_tests.cpp @@ -0,0 +1,175 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + + +// This set of tests checks whether the various settings of the +// --reconfiguration_policy flag behave as expected. + +#include "slave/compatibility.hpp" + +#include <mesos/attributes.hpp> +#include <mesos/resources.hpp> + +#include "tests/mesos.hpp" + +namespace mesos { +namespace internal { +namespace tests { + +class SlaveCompatibilityTest : public MesosTest {}; + + +SlaveInfo createSlaveInfo( + const std::string& resources, + const std::string& attributes) +{ + SlaveID id; + id.set_value("agent"); + + Attributes agentAttributes = Attributes::parse(attributes); + Resources agentResources = Resources::parse(resources).get(); + + SlaveInfo slave; + *(slave.mutable_attributes()) = agentAttributes; + *(slave.mutable_resources()) = agentResources; + *(slave.mutable_id()) = id; + slave.set_hostname(id.value()); + + return slave; +} + + +TEST_F(SlaveCompatibilityTest, Equal) +{ + SlaveInfo original = createSlaveInfo("cpus:500", "foo:bar"); + + SlaveInfo changedAttributes(original); + SlaveInfo changedResources(original); + ASSERT_SOME(slave::compatibility::equal(original, changedAttributes)); + ASSERT_SOME(slave::compatibility::equal(original, changedResources)); + + *(changedAttributes.mutable_attributes()) = Attributes::parse("foo:baz"); + ASSERT_ERROR(slave::compatibility::equal(original, changedAttributes)); + + *(changedResources.mutable_resources()) = Resources::parse("cpus:600").get(); + ASSERT_ERROR(slave::compatibility::equal(original, changedResources)); +} + + +TEST_F(SlaveCompatibilityTest, Additive) +{ + // Changing the hostname is not permitted. + SlaveInfo originalHostname; + originalHostname.set_hostname("host"); + SlaveInfo changedHostname(originalHostname); + ASSERT_SOME(slave::compatibility::additive( + originalHostname, changedHostname)); + + changedHostname.set_hostname("another_host"); + ASSERT_ERROR(slave::compatibility::additive( + originalHostname, changedHostname)); + + // Changing the port is not permitted. + SlaveInfo originalPort; + originalPort.set_port(1234); + SlaveInfo changedPort(originalPort); + ASSERT_SOME(slave::compatibility::additive(originalPort, changedPort)); + + changedPort.set_port(4321); + ASSERT_ERROR(slave::compatibility::additive(originalPort, changedPort)); + + // Resources. + + // Adding new resources is permitted. + SlaveInfo originalResource = createSlaveInfo("cpus:50", ""); + SlaveInfo extendedResource = createSlaveInfo("cpus:50;mem:100", ""); + SlaveInfo modifiedResource = createSlaveInfo("cpus:[100-200]", ""); + ASSERT_SOME(slave::compatibility::additive( + originalResource, extendedResource)); + + // Removing existing resources is not permitted. + ASSERT_ERROR(slave::compatibility::additive( + extendedResource, originalResource)); + + // Changing the type of a resource is not permitted. + ASSERT_ERROR(slave::compatibility::additive( + originalResource, modifiedResource)); + + // Scalar resources can be increased but not decreased. + SlaveInfo originalScalarResource = createSlaveInfo("cpus:50", ""); + SlaveInfo changedScalarResource = createSlaveInfo("cpus:100", ""); + ASSERT_SOME(slave::compatibility::additive( + originalScalarResource, changedScalarResource)); + ASSERT_ERROR(slave::compatibility::additive( + changedScalarResource, originalScalarResource)); + + // Range attributes can be extended but not shrinked. + SlaveInfo originalRangeResource = createSlaveInfo("range:[100-200]", ""); + SlaveInfo changedRangeResource = createSlaveInfo("range:[100-300]", ""); + ASSERT_SOME(slave::compatibility::additive( + originalRangeResource, changedRangeResource)); + ASSERT_ERROR(slave::compatibility::additive( + changedRangeResource, originalRangeResource)); + + // Set attributes can be extended but not shrinked. + SlaveInfo originalSetResource = createSlaveInfo("set:{}", ""); + SlaveInfo changedSetResource = createSlaveInfo("set:{a,b}", ""); + ASSERT_SOME(slave::compatibility::additive( + originalSetResource, changedSetResource)); + ASSERT_ERROR(slave::compatibility::additive( + changedSetResource, originalSetResource)); + + // Attributes. + + // Adding new attributes is permitted. + SlaveInfo originalAttribute = createSlaveInfo("", "os:lucid"); + SlaveInfo extendedAttribute = createSlaveInfo("", "os:lucid;dc:amsterdam"); + SlaveInfo modifiedAttribute = createSlaveInfo("", "os:4"); + ASSERT_SOME(slave::compatibility::additive( + originalAttribute, extendedAttribute)); + + // Removing existing attributes is not permitted. + ASSERT_ERROR(slave::compatibility::additive( + extendedAttribute, originalAttribute)); + + // Changing the type of an attribute is not permitted. + ASSERT_ERROR(slave::compatibility::additive( + originalAttribute, modifiedAttribute)); + + // Changing value of a text attribute is not permitted. + SlaveInfo originalTextAttribute = createSlaveInfo("", "os:lucid"); + SlaveInfo changedTextAttribute = createSlaveInfo("", "os:trusty"); + ASSERT_ERROR(slave::compatibility::additive( + originalTextAttribute, changedTextAttribute)); + + // Changing the value of a scalar attribute is not permitted. + SlaveInfo originalScalarAttribute = createSlaveInfo("", "rack:1"); + SlaveInfo changedScalarAttribute = createSlaveInfo("", "rack:2"); + ASSERT_ERROR(slave::compatibility::additive( + originalScalarAttribute, changedScalarAttribute)); + + // Range attributes can be extended but not shrinked. + SlaveInfo originalRangeAttribute = createSlaveInfo("", "range:[100-200]"); + SlaveInfo changedRangeAttribute = createSlaveInfo("", "range:[100-300]"); + ASSERT_SOME(slave::compatibility::additive( + originalRangeAttribute, changedRangeAttribute)); + ASSERT_ERROR(slave::compatibility::additive( + changedRangeAttribute, originalRangeAttribute)); +} + +} // namespace tests { +} // namespace internal { +} // namespace mesos { http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/tests/slave_recovery_tests.cpp ---------------------------------------------------------------------- diff --git a/src/tests/slave_recovery_tests.cpp b/src/tests/slave_recovery_tests.cpp index 7674e60..253b0fc 100644 --- a/src/tests/slave_recovery_tests.cpp +++ b/src/tests/slave_recovery_tests.cpp @@ -4559,6 +4559,7 @@ TYPED_TEST(SlaveRecoveryTest, RestartBeforeContainerizerLaunch) Try<Owned<cluster::Slave>> slave = this->StartSlave(detector.get(), &containerizer1, flags); + ASSERT_SOME(slave); // Enable checkpointing for the framework. @@ -4637,6 +4638,140 @@ TYPED_TEST(SlaveRecoveryTest, RestartBeforeContainerizerLaunch) } +// This test starts a task, restarts the slave with increased +// resources while the task is still running, and then stops +// the task, verifying that all resources are seen in subsequent +// offers. +TYPED_TEST(SlaveRecoveryTest, AgentReconfigurationWithRunningTask) +{ + // Start a master. + Try<Owned<cluster::Master>> master = this->StartMaster(); + ASSERT_SOME(master); + + // Start a framework. + FrameworkInfo frameworkInfo = DEFAULT_FRAMEWORK_INFO; + frameworkInfo.set_checkpoint(true); + + MockScheduler sched; + MesosSchedulerDriver driver( + &sched, frameworkInfo, master.get()->pid, DEFAULT_CREDENTIAL); + + EXPECT_CALL(sched, registered(_, _, _)); + + Future<vector<Offer>> offers1; + EXPECT_CALL(sched, resourceOffers(_, _)) + .WillOnce(FutureArg<1>(&offers1)) + .WillRepeatedly(Return()); // Ignore subsequent offers. + + driver.start(); + + // Start a slave. + slave::Flags flags = this->CreateSlaveFlags(); + flags.resources = "cpus:5;mem:0;disk:0;ports:0"; + + Fetcher fetcher(flags); + + Try<TypeParam*> _containerizer = TypeParam::create(flags, true, &fetcher); + ASSERT_SOME(_containerizer); + Owned<slave::Containerizer> containerizer(_containerizer.get()); + + Owned<MasterDetector> detector = master.get()->createDetector(); + Try<Owned<cluster::Slave>> slave = + this->StartSlave(detector.get(), containerizer.get(), flags); + + ASSERT_SOME(slave); + + // Start a long-running task on the slave. + AWAIT_READY(offers1); + ASSERT_FALSE(offers1->empty()); + + EXPECT_EQ( + offers1.get()[0].resources(), + allocatedResources(Resources::parse("cpus:5").get(), "*")); + + SlaveID slaveId = offers1.get()[0].slave_id(); + TaskInfo task = createTask( + slaveId, Resources::parse("cpus:3").get(), "sleep 1000"); + + Future<TaskStatus> statusStarting; + Future<TaskStatus> statusRunning; + Future<TaskStatus> statusKilled; + EXPECT_CALL(sched, statusUpdate(_, _)) + .WillOnce(FutureArg<1>(&statusStarting)) + .WillOnce(FutureArg<1>(&statusRunning)) + .WillOnce(FutureArg<1>(&statusKilled)); + + driver.launchTasks(offers1.get()[0].id(), {task}); + + AWAIT_READY(statusStarting); + AWAIT_READY(statusRunning); + + // Grab one of the offers while the task is running. + Future<vector<Offer>> offers2; + EXPECT_CALL(sched, resourceOffers(_, _)) + .WillOnce(FutureArg<1>(&offers2)) + .WillRepeatedly(Return()); // Ignore subsequent offers. + + AWAIT_READY(offers2); + ASSERT_FALSE(offers2->empty()); + + EXPECT_EQ( + offers2.get()[0].resources(), + allocatedResources(Resources::parse("cpus:2").get(), "*")); + + driver.declineOffer(offers2.get()[0].id()); + + // Restart the slave with increased resources. + slave.get()->terminate(); + flags.reconfiguration_policy = "additive"; + flags.resources = "cpus:10;mem:512;disk:0;ports:0"; + + // Restart the slave with a new containerizer. + _containerizer = TypeParam::create(flags, true, &fetcher); + ASSERT_SOME(_containerizer); + containerizer.reset(_containerizer.get()); + + Future<SlaveReregisteredMessage> slaveReregistered = + FUTURE_PROTOBUF(SlaveReregisteredMessage(), _, _); + + // Grab one of the offers after the slave was restarted. + Future<vector<Offer>> offers3; + EXPECT_CALL(sched, resourceOffers(_, _)) + .WillOnce(FutureArg<1>(&offers3)) + .WillRepeatedly(Return()); // Ignore subsequent offers. + + slave = this->StartSlave(detector.get(), containerizer.get(), flags); + ASSERT_SOME(slave); + AWAIT_READY(slaveReregistered); + + AWAIT_READY(offers3); + EXPECT_EQ( + offers3.get()[0].resources(), + allocatedResources(Resources::parse("cpus:7;mem:512").get(), "*")); + + // Decline so we get the resources offered again with the next offer. + driver.declineOffer(offers3.get()[0].id()); + + // Kill the task + driver.killTask(task.task_id()); + AWAIT_READY(statusKilled); + ASSERT_EQ(TASK_KILLED, statusKilled->state()); + + // Grab one of the offers after the task was killed. + Future<vector<Offer>> offers4; + EXPECT_CALL(sched, resourceOffers(_, _)) + .WillOnce(FutureArg<1>(&offers4)) + .WillRepeatedly(Return()); // Ignore subsequent offers. + + EXPECT_EQ( + offers4.get()[0].resources(), + allocatedResources(Resources::parse("cpus:10;mem:512").get(), "*")); + + driver.stop(); + driver.join(); +} + + // We explicitly instantiate a SlaveRecoveryTest for test cases where // we assume we'll only have the MesosContainerizer. class MesosContainerizerSlaveRecoveryTest http://git-wip-us.apache.org/repos/asf/mesos/blob/63e3336b/src/tests/slave_tests.cpp ---------------------------------------------------------------------- diff --git a/src/tests/slave_tests.cpp b/src/tests/slave_tests.cpp index a7f6658..eee6aa4 100644 --- a/src/tests/slave_tests.cpp +++ b/src/tests/slave_tests.cpp @@ -90,6 +90,7 @@ #include "tests/limiter.hpp" #include "tests/mesos.hpp" #include "tests/mock_slave.hpp" +#include "tests/resources_utils.hpp" #include "tests/utils.hpp" using namespace mesos::internal::slave; @@ -8830,6 +8831,103 @@ TEST_F(SlaveTest, ResourceVersions) } +// Test that it is possible to add additional resources, attributes, +// and a domain when the reconfiguration policy is set to +// `additive`. +TEST_F(SlaveTest, ReconfigurationPolicy) +{ + DomainInfo domain = flags::parse<DomainInfo>( + "{" + " \"fault_domain\": {" + " \"region\": {\"name\": \"europe\"}," + " \"zone\": {\"name\": \"europe-b2\"}" + " }" + "}").get(); + + master::Flags masterFlags = CreateMasterFlags(); + // Need to set a master domain, otherwise it will reject a slave with + // a configured domain. + masterFlags.domain = domain; + + Try<Owned<cluster::Master>> master = StartMaster(masterFlags); + ASSERT_SOME(master); + + slave::Flags slaveFlags = CreateSlaveFlags(); + slaveFlags.attributes = "distro:debian"; + slaveFlags.resources = "cpus:4;mem:32;disk:512"; + + MockExecutor exec(DEFAULT_EXECUTOR_ID); + TestContainerizer containerizer(&exec); + Owned<MasterDetector> detector = master.get()->createDetector(); + + // Start a slave. + Future<SlaveRegisteredMessage> slaveRegisteredMessage = + FUTURE_PROTOBUF(SlaveRegisteredMessage(), master.get()->pid, _); + + Try<Owned<cluster::Slave>> slave = + StartSlave(detector.get(), &containerizer, slaveFlags); + + ASSERT_SOME(slave); + + // Wait until the slave registers to ensure that it has successfully + // checkpointed its state. + AWAIT_READY(slaveRegisteredMessage); + + slave.get()->terminate(); + slave->reset(); + + // Do a valid reconfiguration. + slaveFlags.reconfiguration_policy = "additive"; + slaveFlags.resources = "cpus:8;mem:128;disk:512"; + slaveFlags.attributes = "distro:debian;version:8"; + slaveFlags.domain = domain; + + // Restart slave. + Future<SlaveReregisteredMessage> slaveReregisteredMessage = + FUTURE_PROTOBUF(SlaveReregisteredMessage(), master.get()->pid, _); + + slave = StartSlave(detector.get(), &containerizer, slaveFlags); + + ASSERT_SOME(slave); + + // If we get here without the slave exiting, things are working as expected. + AWAIT_READY(slaveReregisteredMessage); + + // Start scheduler and check that it gets offered the updated resources + MockScheduler sched; + MesosSchedulerDriver driver( + &sched, DEFAULT_FRAMEWORK_INFO, master.get()->pid, DEFAULT_CREDENTIAL); + + EXPECT_CALL(sched, registered(&driver, _, _)); + + Future<vector<Offer>> offers; + EXPECT_CALL(sched, resourceOffers(&driver, _)) + .WillOnce(FutureArg<1>(&offers)) + .WillRepeatedly(Return()); // Ignore subsequent offers. + + driver.start(); + + AWAIT_READY(offers); + EXPECT_EQ(1u, offers->size()); + + // Verify that the offer contains the new domain, attributes and resources. + EXPECT_TRUE(offers.get()[0].has_domain()); + EXPECT_EQ( + Attributes(offers.get()[0].attributes()), + Attributes::parse(slaveFlags.attributes.get())); + + // The resources are slightly transformed by both master and slave + // before they end up in an offer (in particular, ports are implicitly + // added and they're assigned to role '*'), so we cannot simply compare + // for equality. + Resources offeredResources = Resources(offers.get()[0].resources()); + Resources reconfiguredResources = allocatedResources( + Resources::parse(slaveFlags.resources.get()).get(), "*"); + + EXPECT_TRUE(offeredResources.contains(reconfiguredResources)); +} + + // This test checks that a resource provider triggers an // `UpdateSlaveMessage` to be sent to the master if an non-speculated // offer operation fails in the resource provider.
