You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by bb...@apache.org on 2018/09/03 12:58:42 UTC
[mesos] 05/06: Made RP manager only send resource provider ID on
state updates.
This is an automated email from the ASF dual-hosted git repository.
bbannier pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit 75bc091e123744003f2ef956be54ad7f562c4815
Author: Benjamin Bannier <bb...@apache.org>
AuthorDate: Wed Aug 15 09:37:12 2018 +0200
Made RP manager only send resource provider ID on state updates.
With the introduction of the resource provider `SUBSCRIBE` event which
contains the full `ResourceProviderInfo` of a subscribed resource
provider, we can avoid sending the `ResourceProviderInfo` in
`UPDATE_STATE` events and instead only send a `ResourceProviderID`
like in most other messages.
Review: https://reviews.apache.org/r/68362
---
src/resource_provider/manager.cpp | 2 +-
src/resource_provider/message.hpp | 4 +-
src/slave/slave.cpp | 153 +++++++++++---------------
src/tests/resource_provider_manager_tests.cpp | 3 +-
4 files changed, 70 insertions(+), 92 deletions(-)
diff --git a/src/resource_provider/manager.cpp b/src/resource_provider/manager.cpp
index 4199e5b..f4a5090 100644
--- a/src/resource_provider/manager.cpp
+++ b/src/resource_provider/manager.cpp
@@ -898,7 +898,7 @@ void ResourceProviderManagerProcess::updateState(
<< resourceProvider->info.id();
ResourceProviderMessage::UpdateState updateState{
- resourceProvider->info,
+ resourceProvider->info.id(),
update.resource_version_uuid(),
update.resources(),
std::move(operations)};
diff --git a/src/resource_provider/message.hpp b/src/resource_provider/message.hpp
index 19e1be1..0acde65 100644
--- a/src/resource_provider/message.hpp
+++ b/src/resource_provider/message.hpp
@@ -71,7 +71,7 @@ struct ResourceProviderMessage
struct UpdateState
{
- ResourceProviderInfo info;
+ ResourceProviderID resourceProviderId;
UUID resourceVersion;
Resources totalResources;
hashmap<UUID, Operation> operations;
@@ -124,7 +124,7 @@ inline std::ostream& operator<<(
CHECK_SOME(updateState);
return stream
- << updateState->info.id() << " "
+ << updateState->resourceProviderId << " "
<< updateState->totalResources;
}
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 4829f00..3bd808e 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -7670,105 +7670,84 @@ void Slave::handleResourceProviderMessage(
const ResourceProviderMessage::UpdateState& updateState =
message->updateState.get();
- CHECK(updateState.info.has_id());
- const ResourceProviderID& resourceProviderId = updateState.info.id();
-
ResourceProvider* resourceProvider =
- getResourceProvider(resourceProviderId);
-
- if (resourceProvider == nullptr) {
- resourceProvider = new ResourceProvider(
- updateState.info,
- updateState.totalResources,
- updateState.resourceVersion);
+ getResourceProvider(updateState.resourceProviderId);
- addResourceProvider(resourceProvider);
-
- foreachvalue (const Operation& operation,
- updateState.operations) {
- addOperation(new Operation(operation));
- }
+ CHECK(resourceProvider);
+ if (resourceProvider->totalResources != updateState.totalResources) {
// Update the 'total' in the Slave.
+ CHECK(totalResources.contains(resourceProvider->totalResources));
+ totalResources -= resourceProvider->totalResources;
totalResources += updateState.totalResources;
- } else {
- // Always update the resource provider info.
- resourceProvider->info = updateState.info;
- if (resourceProvider->totalResources != updateState.totalResources) {
- // Update the 'total' in the Slave.
- CHECK(totalResources.contains(resourceProvider->totalResources));
- totalResources -= resourceProvider->totalResources;
- totalResources += updateState.totalResources;
-
- // Update the 'total' in the resource provider.
- resourceProvider->totalResources = updateState.totalResources;
- }
+ // Update the 'total' in the resource provider.
+ resourceProvider->totalResources = updateState.totalResources;
+ }
- // Update operation state.
- //
- // We only update operations which are not contained in both
- // the known and just received sets. All other operations will
- // be updated via relayed operation status updates.
- const hashset<UUID> knownUuids = resourceProvider->operations.keys();
- const hashset<UUID> receivedUuids = updateState.operations.keys();
-
- // Handle operations known to the agent but not reported by
- // the resource provider. These could be operations where the
- // agent has started tracking an operation, but the resource
- // provider failed over before it could bookkeep the
- // operation.
+ // Update operation state.
+ //
+ // We only update operations which are not contained in both
+ // the known and just received sets. All other operations will
+ // be updated via relayed operation status updates.
+ const hashset<UUID> knownUuids = resourceProvider->operations.keys();
+ const hashset<UUID> receivedUuids = updateState.operations.keys();
+
+ // Handle operations known to the agent but not reported by
+ // the resource provider. These could be operations where the
+ // agent has started tracking an operation, but the resource
+ // provider failed over before it could bookkeep the
+ // operation.
+ //
+ // NOTE: We do not mutate operations statuses here; this would
+ // be the responsibility of an operation status update handler.
+ hashset<UUID> disappearedUuids = knownUuids - receivedUuids;
+ foreach (const UUID& uuid, disappearedUuids) {
+ // TODO(bbannier): Instead of simply dropping an operation
+ // with `removeOperation` here we should instead send a
+ // `Reconcile` message with a failed state to the resource
+ // provider so its status update manager can reliably
+ // deliver the operation status to the framework.
+ removeOperation(resourceProvider->operations.at(uuid));
+ }
+
+ // Handle operations known to the resource provider but not
+ // the agent. This can happen if the agent failed over and the
+ // resource provider reregistered.
+ hashset<UUID> reappearedUuids = receivedUuids - knownUuids;
+ foreach (const UUID& uuid, reappearedUuids) {
+ // Start tracking this operation.
//
- // NOTE: We do not mutate operations statuses here; this would
- // be the responsibility of an operation status update handler.
- hashset<UUID> disappearedUuids = knownUuids - receivedUuids;
- foreach (const UUID& uuid, disappearedUuids) {
- // TODO(bbannier): Instead of simply dropping an operation
- // with `removeOperation` here we should instead send a
- // `Reconcile` message with a failed state to the resource
- // provider so its status update manager can reliably
- // deliver the operation status to the framework.
- removeOperation(resourceProvider->operations.at(uuid));
- }
-
- // Handle operations known to the resource provider but not
- // the agent. This can happen if the agent failed over and the
- // resource provider reregistered.
- hashset<UUID> reappearedUuids = receivedUuids - knownUuids;
- foreach (const UUID& uuid, reappearedUuids) {
- // Start tracking this operation.
- //
- // NOTE: We do not need to update total resources here as its
- // state was sync explicitly with the received total above.
- addOperation(new Operation(updateState.operations.at(uuid)));
- }
+ // NOTE: We do not need to update total resources here as its
+ // state was sync explicitly with the received total above.
+ addOperation(new Operation(updateState.operations.at(uuid)));
+ }
- // Handle operations known to both the agent and the resource provider.
- //
- // If an operation became terminal, its result is already reflected in
- // the total resources reported by the resource provider, and thus it
- // should not be applied again in an operation status update handler
- // when its terminal status update arrives. So we set the terminal
- // `latest_status` here to prevent resource convervions elsewhere.
- //
- // NOTE: We only update the `latest_status` of a known operation if it
- // is not terminal yet here; its `statuses` would be updated by an
- // operation status update handler.
- hashset<UUID> matchedUuids = knownUuids - disappearedUuids;
- foreach (const UUID& uuid, matchedUuids) {
- const Operation& operation = updateState.operations.at(uuid);
- if (operation.has_latest_status() &&
- protobuf::isTerminalState(operation.latest_status().state())) {
- updateOperationLatestStatus(
- getOperation(uuid),
- operation.latest_status());
- }
+ // Handle operations known to both the agent and the resource provider.
+ //
+ // If an operation became terminal, its result is already reflected in
+ // the total resources reported by the resource provider, and thus it
+ // should not be applied again in an operation status update handler
+ // when its terminal status update arrives. So we set the terminal
+ // `latest_status` here to prevent resource convervions elsewhere.
+ //
+ // NOTE: We only update the `latest_status` of a known operation if it
+ // is not terminal yet here; its `statuses` would be updated by an
+ // operation status update handler.
+ hashset<UUID> matchedUuids = knownUuids - disappearedUuids;
+ foreach (const UUID& uuid, matchedUuids) {
+ const Operation& operation = updateState.operations.at(uuid);
+ if (operation.has_latest_status() &&
+ protobuf::isTerminalState(operation.latest_status().state())) {
+ updateOperationLatestStatus(
+ getOperation(uuid),
+ operation.latest_status());
}
-
- // Update resource version of this resource provider.
- resourceProvider->resourceVersion = updateState.resourceVersion;
}
+ // Update resource version of this resource provider.
+ resourceProvider->resourceVersion = updateState.resourceVersion;
+
// Send the updated resources to the master if the agent is running. Note
// that since we have already updated our copy of the latest resource
// provider resources, it is safe to consume this message and wait for the
diff --git a/src/tests/resource_provider_manager_tests.cpp b/src/tests/resource_provider_manager_tests.cpp
index 33b1055..5bb740e 100644
--- a/src/tests/resource_provider_manager_tests.cpp
+++ b/src/tests/resource_provider_manager_tests.cpp
@@ -315,10 +315,9 @@ TEST_P(ResourceProviderManagerHttpApiTest, UpdateState)
AWAIT_READY(message);
EXPECT_EQ(ResourceProviderMessage::Type::UPDATE_STATE, message->type);
- ASSERT_TRUE(message->updateState->info.has_id());
EXPECT_EQ(
devolve(resourceProviderId.get()),
- message->updateState->info.id());
+ message->updateState->resourceProviderId);
EXPECT_EQ(devolve(resources), message->updateState->totalResources);
}
}