| 8016 | |
| 8017 | |
| 8018 | Future<Nothing> Slave::_recoverOperations( |
| 8019 | const Option<state::SlaveState>& state) |
| 8020 | { |
| 8021 | if (state.isNone()) { |
| 8022 | return Nothing(); |
| 8023 | } |
| 8024 | |
| 8025 | operationStatusUpdateManager.initialize( |
| 8026 | defer(self(), &Self::sendOperationStatusUpdate, lambda::_1), |
| 8027 | std::bind( |
| 8028 | &slave::paths::getSlaveOperationUpdatesPath, |
| 8029 | metaDir, |
| 8030 | info.id(), |
| 8031 | lambda::_1)); |
| 8032 | |
| 8033 | if (state->operations.isSome()) { |
| 8034 | foreach (const Operation& operation, state->operations.get()) { |
| 8035 | Result<ResourceProviderID> resourceProviderId = |
| 8036 | getResourceProviderId(operation.info()); |
| 8037 | |
| 8038 | // Only operations affecting agent default resources are checkpointed. |
| 8039 | CHECK(resourceProviderId.isNone()); |
| 8040 | |
| 8041 | addOperation(new Operation(operation)); |
| 8042 | } |
| 8043 | } |
| 8044 | |
| 8045 | // Walk the operation status update streams directories in order to generate |
| 8046 | // the list of streams to recover. |
| 8047 | // |
| 8048 | // NOTE: It is possible for the agent to fail over right after having |
| 8049 | // checkpointed an operation in a `ResourceState` message, but before having |
| 8050 | // created the corresponding stream. |
| 8051 | // |
| 8052 | // In that case the checkpointed message will contain the operation, but the |
| 8053 | // corresponding directory for the operation status update stream will not |
| 8054 | // exist. |
| 8055 | // |
| 8056 | // Since the SUM recovery process will return an error if invoked with an |
| 8057 | // operation ID for which a stream hasn't been created, we can't extract the |
| 8058 | // list of streams to recover from the content of the checkpointed |
| 8059 | // `ResourceState` message. |
| 8060 | Try<list<string>> operationPaths = |
| 8061 | slave::paths::getSlaveOperationPaths(metaDir, info.id()); |
| 8062 | |
| 8063 | if (operationPaths.isError()) { |
| 8064 | return Failure( |
| 8065 | "Failed to find operation status update streams: " + |
| 8066 | operationPaths.error()); |
| 8067 | } |
| 8068 | |
| 8069 | list<id::UUID> operationUuids; |
| 8070 | foreach (const string& path, operationPaths.get()) { |
| 8071 | Try<id::UUID> uuid = |
| 8072 | slave::paths::parseSlaveOperationPath(metaDir, info.id(), path); |
| 8073 | |
| 8074 | if (uuid.isError()) { |
| 8075 | return Failure( |
nothing calls this directly
no test coverage detected