| // 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 "resource_provider/manager.hpp" |
| |
| #include <string> |
| #include <utility> |
| #include <vector> |
| |
| #include <glog/logging.h> |
| |
| #include <mesos/http.hpp> |
| |
| #include <mesos/resource_provider/resource_provider.hpp> |
| |
| #include <mesos/v1/resource_provider/resource_provider.hpp> |
| |
| #include <process/collect.hpp> |
| #include <process/dispatch.hpp> |
| #include <process/id.hpp> |
| #include <process/process.hpp> |
| |
| #include <process/metrics/counter.hpp> |
| #include <process/metrics/metrics.hpp> |
| #include <process/metrics/pull_gauge.hpp> |
| |
| #include <stout/hashmap.hpp> |
| #include <stout/nothing.hpp> |
| #include <stout/protobuf.hpp> |
| #include <stout/uuid.hpp> |
| |
| #include "common/http.hpp" |
| #include "common/protobuf_utils.hpp" |
| #include "common/recordio.hpp" |
| #include "common/resources_utils.hpp" |
| |
| #include "internal/devolve.hpp" |
| #include "internal/evolve.hpp" |
| |
| #include "resource_provider/registry.hpp" |
| #include "resource_provider/validation.hpp" |
| |
| namespace http = process::http; |
| |
| using std::string; |
| using std::vector; |
| |
| using mesos::internal::resource_provider::validation::call::validate; |
| |
| using mesos::resource_provider::AdmitResourceProvider; |
| using mesos::resource_provider::Call; |
| using mesos::resource_provider::Event; |
| using mesos::resource_provider::Registrar; |
| using mesos::resource_provider::RemoveResourceProvider; |
| |
| using mesos::resource_provider::registry::Registry; |
| |
| using process::Failure; |
| using process::Future; |
| using process::Owned; |
| using process::Process; |
| using process::ProcessBase; |
| using process::Promise; |
| using process::Queue; |
| |
| using process::collect; |
| using process::dispatch; |
| using process::spawn; |
| using process::terminate; |
| using process::wait; |
| |
| using process::http::Accepted; |
| using process::http::BadRequest; |
| using process::http::MethodNotAllowed; |
| using process::http::NotAcceptable; |
| using process::http::NotImplemented; |
| using process::http::OK; |
| using process::http::Pipe; |
| using process::http::UnsupportedMediaType; |
| |
| using process::http::authentication::Principal; |
| |
| using process::metrics::Counter; |
| using process::metrics::PullGauge; |
| |
| namespace mesos { |
| namespace internal { |
| |
| mesos::resource_provider::registry::ResourceProvider |
| createRegistryResourceProvider(const ResourceProviderInfo& resourceProviderInfo) |
| { |
| mesos::resource_provider::registry::ResourceProvider resourceProvider; |
| |
| CHECK(resourceProviderInfo.has_id()); |
| resourceProvider.mutable_id()->CopyFrom(resourceProviderInfo.id()); |
| |
| resourceProvider.set_name(resourceProviderInfo.name()); |
| resourceProvider.set_type(resourceProviderInfo.type()); |
| |
| return resourceProvider; |
| } |
| |
| |
| struct ResourceProvider |
| { |
| ResourceProvider( |
| const ResourceProviderInfo& _info, |
| const StreamingHttpConnection<v1::resource_provider::Event>& _http) |
| : info(_info), |
| http(_http) {} |
| |
| ~ResourceProvider() |
| { |
| LOG(INFO) << "Terminating resource provider " << info.id(); |
| |
| http.close(); |
| |
| foreachvalue (const Owned<Promise<Nothing>>& publish, publishes) { |
| publish->fail( |
| "Failed to publish resources from resource provider " + |
| stringify(info.id()) + ": Connection closed"); |
| } |
| } |
| |
| ResourceProviderInfo info; |
| StreamingHttpConnection<v1::resource_provider::Event> http; |
| hashmap<id::UUID, Owned<Promise<Nothing>>> publishes; |
| }; |
| |
| |
| class ResourceProviderManagerProcess |
| : public Process<ResourceProviderManagerProcess> |
| { |
| public: |
| ResourceProviderManagerProcess(Owned<Registrar> _registrar); |
| |
| Future<http::Response> api( |
| const http::Request& request, |
| const Option<Principal>& principal); |
| |
| void applyOperation(const ApplyOperationMessage& message); |
| |
| void acknowledgeOperationStatus( |
| const AcknowledgeOperationStatusMessage& message); |
| |
| void reconcileOperations(const ReconcileOperationsMessage& message); |
| |
| Future<Nothing> removeResourceProvider( |
| const ResourceProviderID& resourceProviderId); |
| |
| Future<Nothing> publishResources(const Resources& resources); |
| |
| Queue<ResourceProviderMessage> messages; |
| |
| private: |
| void subscribe( |
| const StreamingHttpConnection<v1::resource_provider::Event>& http, |
| const Call::Subscribe& subscribe); |
| |
| void _subscribe( |
| const Future<bool>& admitResourceProvider, |
| Owned<ResourceProvider> resourceProvider); |
| |
| void updateOperationStatus( |
| ResourceProvider* resourceProvider, |
| const Call::UpdateOperationStatus& update); |
| |
| ResourceProviderMessage createUpdateOperationStatus( |
| const OperationStatus& operationStatus, |
| const Option<id::UUID>& operationUuid, |
| const Option<FrameworkID>& frameworkId, |
| const Option<OperationStatus>& latestStatus); |
| |
| void updateState( |
| ResourceProvider* resourceProvider, |
| const Call::UpdateState& update); |
| |
| void updatePublishResourcesStatus( |
| ResourceProvider* resourceProvider, |
| const Call::UpdatePublishResourcesStatus& update); |
| |
| Future<Nothing> recover( |
| const mesos::resource_provider::registry::Registry& registry); |
| |
| void initialize() override; |
| |
| ResourceProviderID newResourceProviderId(); |
| |
| double gaugeSubscribed(); |
| |
| struct ResourceProviders |
| { |
| hashmap<ResourceProviderID, Owned<ResourceProvider>> subscribed; |
| hashmap< |
| ResourceProviderID, |
| mesos::resource_provider::registry::ResourceProvider> |
| known; |
| } resourceProviders; |
| |
| struct Metrics |
| { |
| explicit Metrics(const ResourceProviderManagerProcess& manager); |
| ~Metrics(); |
| |
| PullGauge subscribed; |
| Counter subscriptions; |
| Counter disconnections; |
| }; |
| |
| Owned<Registrar> registrar; |
| Promise<Nothing> recovered; |
| |
| Metrics metrics; |
| }; |
| |
| |
| ResourceProviderManagerProcess::ResourceProviderManagerProcess( |
| Owned<Registrar> _registrar) |
| : ProcessBase(process::ID::generate("resource-provider-manager")), |
| registrar(std::move(_registrar)), |
| metrics(*this) |
| { |
| CHECK_NOTNULL(registrar.get()); |
| } |
| |
| |
| void ResourceProviderManagerProcess::initialize() |
| { |
| // Recover the registrar. |
| registrar->recover() |
| .then(defer(self(), &ResourceProviderManagerProcess::recover, lambda::_1)) |
| .onAny([](const Future<Nothing>& recovered) { |
| if (!recovered.isReady()) { |
| LOG(FATAL) |
| << "Failed to recover resource provider manager registry: " |
| << recovered; |
| } |
| }); |
| } |
| |
| |
| Future<Nothing> ResourceProviderManagerProcess::recover( |
| const mesos::resource_provider::registry::Registry& registry) |
| { |
| foreach ( |
| const mesos::resource_provider::registry::ResourceProvider& |
| resourceProvider, |
| registry.resource_providers()) { |
| resourceProviders.known.put(resourceProvider.id(), resourceProvider); |
| } |
| |
| recovered.set(Nothing()); |
| |
| return Nothing(); |
| } |
| |
| |
| Future<http::Response> ResourceProviderManagerProcess::api( |
| const http::Request& request, |
| const Option<Principal>& principal) |
| { |
| // TODO(bbannier): This implementation does not limit the number of messages |
| // in the actor's inbox which could become large should a big number of |
| // resource providers attempt to subscribe before recovery completed. Consider |
| // rejecting requests until the resource provider manager has recovered. This |
| // would likely require implementing retry logic in resource providers. |
| return recovered.future().then(defer( |
| self(), [this, request, principal](const Nothing&) -> http::Response { |
| if (request.method != "POST") { |
| return MethodNotAllowed({"POST"}, request.method); |
| } |
| |
| v1::resource_provider::Call v1Call; |
| |
| // TODO(anand): Content type values are case-insensitive. |
| Option<string> contentType = request.headers.get("Content-Type"); |
| |
| if (contentType.isNone()) { |
| return BadRequest("Expecting 'Content-Type' to be present"); |
| } |
| |
| if (contentType.get() == APPLICATION_PROTOBUF) { |
| if (!v1Call.ParseFromString(request.body)) { |
| return BadRequest("Failed to parse body into Call protobuf"); |
| } |
| } else if (contentType.get() == APPLICATION_JSON) { |
| Try<JSON::Value> value = JSON::parse(request.body); |
| if (value.isError()) { |
| return BadRequest( |
| "Failed to parse body into JSON: " + value.error()); |
| } |
| |
| Try<v1::resource_provider::Call> parse = |
| ::protobuf::parse<v1::resource_provider::Call>(value.get()); |
| |
| if (parse.isError()) { |
| return BadRequest( |
| "Failed to convert JSON into Call protobuf: " + parse.error()); |
| } |
| |
| v1Call = parse.get(); |
| } else { |
| return UnsupportedMediaType( |
| string("Expecting 'Content-Type' of ") + APPLICATION_JSON + |
| " or " + APPLICATION_PROTOBUF); |
| } |
| |
| Call call = devolve(v1Call); |
| |
| Option<Error> error = validate(call, None()); |
| if (error.isSome()) { |
| return BadRequest( |
| "Failed to validate resource_provider::Call: " + error->message); |
| } |
| |
| if (call.type() == Call::SUBSCRIBE) { |
| // We default to JSON 'Content-Type' in the response since an empty |
| // 'Accept' header results in all media types considered acceptable. |
| ContentType acceptType = ContentType::JSON; |
| |
| if (request.acceptsMediaType(APPLICATION_JSON)) { |
| acceptType = ContentType::JSON; |
| } else if (request.acceptsMediaType(APPLICATION_PROTOBUF)) { |
| acceptType = ContentType::PROTOBUF; |
| } else { |
| return NotAcceptable( |
| string("Expecting 'Accept' to allow ") + "'" + |
| APPLICATION_PROTOBUF + "' or '" + APPLICATION_JSON + "'"); |
| } |
| |
| if (request.headers.contains("Mesos-Stream-Id")) { |
| return BadRequest( |
| "Subscribe calls should not include the 'Mesos-Stream-Id' " |
| "header"); |
| } |
| |
| Pipe pipe; |
| OK ok; |
| |
| ok.headers["Content-Type"] = stringify(acceptType); |
| ok.type = http::Response::PIPE; |
| ok.reader = pipe.reader(); |
| |
| // Generate a stream ID and return it in the response. |
| id::UUID streamId = id::UUID::random(); |
| ok.headers["Mesos-Stream-Id"] = streamId.toString(); |
| |
| StreamingHttpConnection<v1::resource_provider::Event> http( |
| pipe.writer(), acceptType, streamId); |
| |
| this->subscribe(http, call.subscribe()); |
| |
| return std::move(ok); |
| } |
| |
| if (!this->resourceProviders.subscribed.contains( |
| call.resource_provider_id())) { |
| return BadRequest("Resource provider is not subscribed"); |
| } |
| |
| ResourceProvider* resourceProvider = CHECK_NOTNULL( |
| this->resourceProviders.subscribed.at(call.resource_provider_id()) |
| .get()); |
| |
| // Perform additional validation with the now known provider info. |
| error = validate(call, resourceProvider->info); |
| if (error.isSome()) { |
| return BadRequest("Inconsistent call: " + error->message); |
| } |
| |
| // This isn't a `SUBSCRIBE` call, so the request should include a stream |
| // ID. |
| if (!request.headers.contains("Mesos-Stream-Id")) { |
| return BadRequest( |
| "All non-subscribe calls should include to 'Mesos-Stream-Id' " |
| "header"); |
| } |
| |
| const string& streamId = request.headers.at("Mesos-Stream-Id"); |
| if (streamId != resourceProvider->http.streamId.toString()) { |
| return BadRequest( |
| "The stream ID '" + streamId + |
| "' included in this request " |
| "didn't match the stream ID currently associated with " |
| " resource provider ID " + |
| resourceProvider->info.id().value()); |
| } |
| |
| switch (call.type()) { |
| case Call::UNKNOWN: { |
| return NotImplemented(); |
| } |
| |
| case Call::SUBSCRIBE: { |
| // `SUBSCRIBE` call should have been handled above. |
| LOG(FATAL) << "Unexpected 'SUBSCRIBE' call"; |
| } |
| |
| case Call::UPDATE_OPERATION_STATUS: { |
| this->updateOperationStatus( |
| resourceProvider, call.update_operation_status()); |
| |
| return Accepted(); |
| } |
| |
| case Call::UPDATE_STATE: { |
| this->updateState(resourceProvider, call.update_state()); |
| return Accepted(); |
| } |
| |
| case Call::UPDATE_PUBLISH_RESOURCES_STATUS: { |
| this->updatePublishResourcesStatus( |
| resourceProvider, call.update_publish_resources_status()); |
| return Accepted(); |
| } |
| } |
| |
| UNREACHABLE(); |
| })); |
| } |
| |
| |
| void ResourceProviderManagerProcess::applyOperation( |
| const ApplyOperationMessage& message) |
| { |
| const Offer::Operation& operation = message.operation_info(); |
| const Option<FrameworkID> frameworkId = message.has_framework_id() |
| ? message.framework_id() |
| : Option<FrameworkID>::none(); |
| const UUID& operationUUID = message.operation_uuid(); |
| |
| Result<ResourceProviderID> resourceProviderId = |
| getResourceProviderId(operation); |
| |
| if (!resourceProviderId.isSome()) { |
| LOG(ERROR) << "Failed to get the resource provider ID of operation " |
| << "'" << operation.id() << "' (uuid: " << operationUUID |
| << ") from " |
| << (frameworkId.isSome() |
| ? "framework " + stringify(frameworkId.get()) |
| : "an operator API call") |
| << ": " |
| << (resourceProviderId.isError() ? resourceProviderId.error() |
| : "Not found"); |
| return; |
| } |
| |
| if (!resourceProviders.subscribed.contains(resourceProviderId.get())) { |
| LOG(WARNING) << "Dropping operation '" << operation.id() << "' (uuid: " |
| << operationUUID << ") from " |
| << (frameworkId.isSome() |
| ? "framework " + stringify(frameworkId.get()) |
| : "an operator API call") |
| << " because resource provider " << resourceProviderId.get() |
| << " is not subscribed"; |
| return; |
| } |
| |
| ResourceProvider* resourceProvider = |
| resourceProviders.subscribed.at(resourceProviderId.get()).get(); |
| |
| CHECK(message.resource_version_uuid().has_resource_provider_id()); |
| |
| CHECK_EQ(message.resource_version_uuid().resource_provider_id(), |
| resourceProviderId.get()) |
| << "Resource provider ID " |
| << message.resource_version_uuid().resource_provider_id() |
| << " in resource version UUID does not match that in the operation " |
| << resourceProviderId.get(); |
| |
| Event event; |
| event.set_type(Event::APPLY_OPERATION); |
| if (frameworkId.isSome()) { |
| event.mutable_apply_operation() |
| ->mutable_framework_id()->CopyFrom(frameworkId.get()); |
| } |
| event.mutable_apply_operation()->mutable_info()->CopyFrom(operation); |
| event.mutable_apply_operation() |
| ->mutable_operation_uuid()->CopyFrom(message.operation_uuid()); |
| event.mutable_apply_operation() |
| ->mutable_resource_version_uuid() |
| ->CopyFrom(message.resource_version_uuid().uuid()); |
| |
| if (!resourceProvider->http.send(event)) { |
| LOG(WARNING) << "Failed to send operation '" << operation.id() << "' " |
| << "(uuid: " << operationUUID << ") from " |
| << (frameworkId.isSome() |
| ? "framework " + stringify(frameworkId.get()) |
| : "an operator API call") |
| << " to resource provider " << resourceProviderId.get() |
| << ": connection closed"; |
| } |
| } |
| |
| |
| void ResourceProviderManagerProcess::acknowledgeOperationStatus( |
| const AcknowledgeOperationStatusMessage& message) |
| { |
| CHECK(message.has_resource_provider_id()); |
| |
| if (!resourceProviders.subscribed.contains(message.resource_provider_id())) { |
| LOG(WARNING) << "Dropping operation status acknowledgement with" |
| << " status_uuid " << message.status_uuid() << " and" |
| << " operation_uuid " << message.operation_uuid() << " because" |
| << " resource provider " << message.resource_provider_id() |
| << " is not subscribed"; |
| return; |
| } |
| |
| ResourceProvider& resourceProvider = |
| *resourceProviders.subscribed.at(message.resource_provider_id()); |
| |
| Event event; |
| event.set_type(Event::ACKNOWLEDGE_OPERATION_STATUS); |
| event.mutable_acknowledge_operation_status() |
| ->mutable_status_uuid() |
| ->CopyFrom(message.status_uuid()); |
| event.mutable_acknowledge_operation_status() |
| ->mutable_operation_uuid() |
| ->CopyFrom(message.operation_uuid()); |
| |
| if (!resourceProvider.http.send(event)) { |
| LOG(WARNING) << "Failed to send operation status acknowledgement with" |
| << " status_uuid " << message.status_uuid() << " and" |
| << " operation_uuid " << message.operation_uuid() << " to" |
| << " resource provider " << message.resource_provider_id() |
| << ": connection closed"; |
| } |
| } |
| |
| |
| void ResourceProviderManagerProcess::reconcileOperations( |
| const ReconcileOperationsMessage& message) |
| { |
| hashmap<ResourceProviderID, Event> events; |
| |
| auto addOperation = |
| [&events](const ReconcileOperationsMessage::Operation& operation) { |
| const ResourceProviderID resourceProviderId = |
| operation.resource_provider_id(); |
| |
| if (events.contains(resourceProviderId)) { |
| events.at(resourceProviderId).mutable_reconcile_operations() |
| ->add_operation_uuids()->CopyFrom(operation.operation_uuid()); |
| } else { |
| Event event; |
| event.set_type(Event::RECONCILE_OPERATIONS); |
| event.mutable_reconcile_operations() |
| ->add_operation_uuids()->CopyFrom(operation.operation_uuid()); |
| |
| events[resourceProviderId] = event; |
| } |
| }; |
| |
| // If the framework ID is set, then this reconciliation request originated |
| // from the framework; otherwise, it originated from the master. If it's from |
| // the framework, then we return an operation state based on whether or not |
| // the resource provider is known. If it's from the master, then we forward it |
| // to the resource provider to determine whether or not this operation was |
| // dropped en route to the agent. |
| Option<FrameworkID> frameworkId = message.has_framework_id() |
| ? message.framework_id() |
| : Option<FrameworkID>::none(); |
| |
| // Construct events for individual resource providers, or send appropriate |
| // operation status updates if resource providers are not subscribed. |
| foreach ( |
| const ReconcileOperationsMessage::Operation& operation, |
| message.operations()) { |
| if (operation.has_resource_provider_id()) { |
| if (!resourceProviders.subscribed.contains( |
| operation.resource_provider_id())) { |
| OperationState state; |
| if (resourceProviders.known.contains( |
| operation.resource_provider_id())) { |
| // The resource provider is not currently subscribed, but since it is |
| // known, we expect it to resubscribe. We send OPERATION_RECOVERING in |
| // this case. |
| state = OPERATION_RECOVERING; |
| } else { |
| // The resource provider is not known; we return OPERATION_UNKNOWN in |
| // this case. |
| state = OPERATION_UNKNOWN; |
| } |
| |
| Option<id::UUID> uuid; |
| if (operation.has_operation_uuid()) { |
| Try<id::UUID> uuid_ = |
| id::UUID::fromBytes(operation.operation_uuid().value()); |
| CHECK_SOME(uuid_); |
| uuid = uuid_.get(); |
| } |
| |
| // NOTE: the `SlaveID` is left blank in the operation status below |
| // because the RP manager does not have access to this information. The |
| // agent will inject the ID before forwarding to the master. |
| messages.put( |
| createUpdateOperationStatus( |
| protobuf::createOperationStatus( |
| state, |
| operation.has_operation_id() |
| ? operation.operation_id() |
| : Option<OperationID>::none(), |
| None(), |
| None(), |
| uuid, |
| None(), |
| operation.resource_provider_id()), |
| uuid, |
| frameworkId, |
| None())); |
| |
| continue; |
| } |
| |
| addOperation(operation); |
| } |
| } |
| |
| foreachpair ( |
| const ResourceProviderID& resourceProviderId, |
| const Event& event, |
| events) { |
| CHECK(resourceProviders.subscribed.contains(resourceProviderId)); |
| ResourceProvider& resourceProvider = |
| *resourceProviders.subscribed.at(resourceProviderId); |
| |
| if (!resourceProvider.http.send(event)) { |
| LOG(WARNING) << "Failed to send operation reconciliation event" |
| << " to resource provider " << resourceProviderId |
| << ": connection closed"; |
| } |
| } |
| } |
| |
| |
| Future<Nothing> ResourceProviderManagerProcess::removeResourceProvider( |
| const ResourceProviderID& resourceProviderId) |
| { |
| LOG(INFO) << "Removing resource provider " << resourceProviderId; |
| |
| Future<bool> removeResourceProvider = registrar |
| ->apply(Owned<mesos::resource_provider::Registrar::Operation>( |
| new RemoveResourceProvider(resourceProviderId))); |
| |
| removeResourceProvider.onAny( |
| [resourceProviderId](const Future<bool>& removeResourceProvider) { |
| if (!removeResourceProvider.isReady()) { |
| LOG(ERROR) << "Not removing resource provider " << resourceProviderId |
| << " as registry update did not succeed: " |
| << removeResourceProvider; |
| } |
| }); |
| |
| return removeResourceProvider |
| .then(defer( |
| self(), |
| [this, resourceProviderId](const Future<bool>& removeResourceProvider) { |
| // TODO(bbannier): We should also on a best effort basis send a |
| // `TEARDOWN` event when a removed resource provider attempts to |
| // resubscribe. This can happen e.g., if the event gets lost on its |
| // way to the resource provider, or if the resource provider is not |
| // connected. |
| if (resourceProviders.subscribed.contains(resourceProviderId)) { |
| const Owned<ResourceProvider>& resourceProvider = |
| resourceProviders.subscribed.at(resourceProviderId); |
| Event event; |
| event.set_type(Event::TEARDOWN); |
| |
| if (!resourceProvider->http.send(event)) { |
| LOG(WARNING) |
| << "Failed to send TEARDOWN event to resource provider " |
| << resourceProviderId << ": connection closed"; |
| } |
| } else { |
| LOG(WARNING) |
| << "Failed to send TEARDOWN event to resource provider " |
| << resourceProviderId << ": resource provider not subscribed"; |
| } |
| |
| resourceProviders.known.erase(resourceProviderId); |
| resourceProviders.subscribed.erase(resourceProviderId); |
| |
| ResourceProviderMessage::Remove remove{resourceProviderId}; |
| |
| ResourceProviderMessage message; |
| message.type = ResourceProviderMessage::Type::REMOVE; |
| message.remove = std::move(remove); |
| |
| messages.put(std::move(message)); |
| |
| return Nothing(); |
| })); |
| } |
| |
| |
| Future<Nothing> ResourceProviderManagerProcess::publishResources( |
| const Resources& resources) |
| { |
| hashmap<ResourceProviderID, Resources> providedResources; |
| |
| foreach (const Resource& resource, resources) { |
| // NOTE: We ignore agent default resources here because those |
| // resources do not need publish, and shouldn't be handled by the |
| // resource provider manager. |
| if (!resource.has_provider_id()) { |
| continue; |
| } |
| |
| const ResourceProviderID& resourceProviderId = resource.provider_id(); |
| |
| if (!resourceProviders.subscribed.contains(resourceProviderId)) { |
| // TODO(chhsiao): If the manager is running on an agent and the |
| // resource comes from an external resource provider, we may want |
| // to load the provider's agent component. |
| return Failure( |
| "Resource provider " + stringify(resourceProviderId) + |
| " is not subscribed"); |
| } |
| |
| providedResources[resourceProviderId] += resource; |
| } |
| |
| vector<Future<Nothing>> futures; |
| |
| foreachpair (const ResourceProviderID& resourceProviderId, |
| const Resources& resources, |
| providedResources) { |
| id::UUID uuid = id::UUID::random(); |
| |
| Event event; |
| event.set_type(Event::PUBLISH_RESOURCES); |
| *event.mutable_publish_resources()->mutable_uuid() = |
| protobuf::createUUID(uuid); |
| *event.mutable_publish_resources()->mutable_resources() = resources; |
| |
| ResourceProvider* resourceProvider = |
| resourceProviders.subscribed.at(resourceProviderId).get(); |
| |
| LOG(INFO) |
| << "Sending PUBLISH event " << uuid << " with resources '" << resources |
| << "' to resource provider " << resourceProviderId; |
| |
| if (!resourceProvider->http.send(event)) { |
| return Failure( |
| "Failed to send PUBLISH_RESOURCES event to resource provider " + |
| stringify(resourceProviderId) + ": connection closed"); |
| } |
| |
| Owned<Promise<Nothing>> promise(new Promise<Nothing>()); |
| futures.push_back(promise->future()); |
| resourceProvider->publishes.put(uuid, std::move(promise)); |
| } |
| |
| return collect(futures).then([] { return Nothing(); }); |
| } |
| |
| |
| void ResourceProviderManagerProcess::subscribe( |
| const StreamingHttpConnection<v1::resource_provider::Event>& http, |
| const Call::Subscribe& subscribe) |
| { |
| const ResourceProviderInfo& resourceProviderInfo = |
| subscribe.resource_provider_info(); |
| |
| LOG(INFO) << "Subscribing resource provider " << resourceProviderInfo; |
| |
| // We always create a new `ResourceProvider` struct when a |
| // resource provider subscribes or resubscribes, and replace the |
| // existing `ResourceProvider` if needed. |
| Owned<ResourceProvider> resourceProvider( |
| new ResourceProvider(resourceProviderInfo, http)); |
| |
| Future<bool> admitResourceProvider; |
| |
| if (!resourceProviderInfo.has_id()) { |
| // The resource provider is subscribing for the first time. |
| resourceProvider->info.mutable_id()->CopyFrom(newResourceProviderId()); |
| |
| // If we are handing out a new `ResourceProviderID` persist the ID by |
| // triggering a `AdmitResourceProvider` operation on the registrar. |
| admitResourceProvider = |
| registrar->apply(Owned<mesos::resource_provider::Registrar::Operation>( |
| new AdmitResourceProvider( |
| createRegistryResourceProvider(resourceProvider->info)))); |
| } else { |
| // TODO(chhsiao): The resource provider is resubscribing after being |
| // restarted or an agent failover. The 'ResourceProviderInfo' might |
| // have been updated, but its type and name should remain the same. |
| // We should checkpoint its 'type', 'name' and ID, then check if the |
| // resubscription is consistent with the checkpointed record. |
| |
| const ResourceProviderID& resourceProviderId = resourceProviderInfo.id(); |
| |
| if (!resourceProviders.known.contains(resourceProviderId)) { |
| LOG(INFO) |
| << "Dropping resubscription attempt of resource provider with ID " |
| << resourceProviderId |
| << " since it is unknown"; |
| |
| return; |
| } |
| |
| // Check whether the resource provider has change |
| // information which should be static. |
| mesos::resource_provider::registry::ResourceProvider resourceProvider_ = |
| createRegistryResourceProvider(resourceProvider->info); |
| |
| const mesos::resource_provider::registry::ResourceProvider& |
| storedResourceProvider = resourceProviders.known.at(resourceProviderId); |
| |
| if (resourceProvider_ != storedResourceProvider) { |
| LOG(INFO) |
| << "Dropping resubscription attempt of resource provider " |
| << resourceProvider_ |
| << " since it does not match the previous information " |
| << storedResourceProvider; |
| return; |
| } |
| |
| // If the resource provider is known we do not need to admit it |
| // again, and the registrar operation implicitly succeeded. |
| admitResourceProvider = true; |
| } |
| |
| admitResourceProvider.onAny(defer( |
| self(), |
| &ResourceProviderManagerProcess::_subscribe, |
| lambda::_1, |
| std::move(resourceProvider))); |
| } |
| |
| |
| void ResourceProviderManagerProcess::_subscribe( |
| const Future<bool>& admitResourceProvider, |
| Owned<ResourceProvider> resourceProvider) |
| { |
| if (!admitResourceProvider.isReady()) { |
| LOG(INFO) |
| << "Not subscribing resource provider " << resourceProvider->info.id() |
| << " as registry update did not succeed: " << admitResourceProvider; |
| |
| return; |
| } |
| |
| CHECK(admitResourceProvider.get()) |
| << "Could not admit resource provider " << resourceProvider->info.id() |
| << " as registry update was rejected"; |
| |
| const ResourceProviderID& resourceProviderId = resourceProvider->info.id(); |
| |
| resourceProvider->http.closed() |
| .onAny(defer(self(), [=](const Future<Nothing>& future) { |
| // Iff the remote side closes the HTTP connection, the future will be |
| // ready. We will remove the resource provider in that case. |
| // This side closes the HTTP connection only when removing a resource |
| // provider, therefore we shouldn't try to remove it again here. |
| if (future.isReady()) { |
| CHECK(resourceProviders.subscribed.contains(resourceProviderId)); |
| |
| // NOTE: All pending futures of publish requests for the resource |
| // provider will become failed. |
| resourceProviders.subscribed.erase(resourceProviderId); |
| } |
| |
| ResourceProviderMessage::Disconnect disconnect{resourceProviderId}; |
| |
| ResourceProviderMessage message; |
| message.type = ResourceProviderMessage::Type::DISCONNECT; |
| message.disconnect = std::move(disconnect); |
| |
| messages.put(std::move(message)); |
| |
| ++metrics.disconnections; |
| })); |
| |
| if (!resourceProviders.known.contains(resourceProviderId)) { |
| mesos::resource_provider::registry::ResourceProvider resourceProvider_ = |
| createRegistryResourceProvider(resourceProvider->info); |
| |
| resourceProviders.known.put( |
| resourceProviderId, |
| std::move(resourceProvider_)); |
| } |
| |
| ResourceProviderMessage::Subscribe subscribe{resourceProvider->info}; |
| |
| ResourceProviderMessage message; |
| message.type = ResourceProviderMessage::Type::SUBSCRIBE; |
| message.subscribe = std::move(subscribe); |
| |
| // Store a pointer to the resource provider so that we can access it |
| // even after moving it into the list of subscribed providers below. |
| ResourceProvider* _resourceProvider = resourceProvider.get(); |
| |
| // TODO(jieyu): Start heartbeat for the resource provider. |
| resourceProviders.subscribed.put( |
| resourceProviderId, |
| std::move(resourceProvider)); |
| |
| messages.put(std::move(message)); |
| ++metrics.subscriptions; |
| |
| // Only send a `SUBSCRIBED` event after we have updated our internal |
| // bookkeeping so that e.g., metrics always reflect correct |
| // subscription state. |
| Event event; |
| event.set_type(Event::SUBSCRIBED); |
| event.mutable_subscribed()->mutable_provider_id() |
| ->CopyFrom(resourceProviderId); |
| |
| if (!_resourceProvider->http.send(event)) { |
| LOG(WARNING) << "Failed to send SUBSCRIBED event to resource provider " |
| << resourceProviderId << ": connection closed"; |
| return; |
| } |
| } |
| |
| |
| void ResourceProviderManagerProcess::updateOperationStatus( |
| ResourceProvider* resourceProvider, |
| const Call::UpdateOperationStatus& update) |
| { |
| CHECK_EQ(update.status().resource_provider_id(), resourceProvider->info.id()); |
| |
| Option<FrameworkID> frameworkId; |
| if (update.has_framework_id()) { |
| frameworkId = update.framework_id(); |
| } |
| |
| Option<OperationStatus> latestStatus; |
| if (update.has_latest_status()) { |
| CHECK_EQ( |
| update.latest_status().resource_provider_id(), |
| resourceProvider->info.id()); |
| |
| latestStatus = update.latest_status(); |
| } |
| |
| Try<id::UUID> uuid = id::UUID::fromBytes(update.operation_uuid().value()); |
| CHECK_SOME(uuid); |
| |
| messages.put( |
| createUpdateOperationStatus( |
| update.status(), |
| uuid.get(), |
| frameworkId, |
| latestStatus)); |
| } |
| |
| |
| ResourceProviderMessage |
| ResourceProviderManagerProcess::createUpdateOperationStatus( |
| const OperationStatus& operationStatus, |
| const Option<id::UUID>& operationUuid, |
| const Option<FrameworkID>& frameworkId, |
| const Option<OperationStatus>& latestStatus) |
| { |
| ResourceProviderMessage::UpdateOperationStatus body; |
| body.update.mutable_status()->CopyFrom(operationStatus); |
| |
| if (operationUuid.isSome()) { |
| body.update.mutable_operation_uuid()->CopyFrom( |
| protobuf::createUUID(operationUuid.get())); |
| } |
| |
| if (frameworkId.isSome()) { |
| body.update.mutable_framework_id()->CopyFrom(frameworkId.get()); |
| } |
| |
| if (latestStatus.isSome()) { |
| body.update.mutable_latest_status()->CopyFrom(latestStatus.get()); |
| } |
| |
| ResourceProviderMessage message; |
| message.type = ResourceProviderMessage::Type::UPDATE_OPERATION_STATUS; |
| message.updateOperationStatus = std::move(body); |
| |
| return message; |
| } |
| |
| |
| void ResourceProviderManagerProcess::updateState( |
| ResourceProvider* resourceProvider, |
| const Call::UpdateState& update) |
| { |
| foreach (const Resource& resource, update.resources()) { |
| CHECK_EQ(resource.provider_id(), resourceProvider->info.id()); |
| } |
| |
| // TODO(chhsiao): Report pending operations. |
| |
| hashmap<UUID, Operation> operations; |
| foreach (const Operation &operation, update.operations()) { |
| operations.put(operation.uuid(), operation); |
| } |
| |
| LOG(INFO) |
| << "Received UPDATE_STATE call with resources '" << update.resources() |
| << "' and " << operations.size() << " operations from resource provider " |
| << resourceProvider->info.id(); |
| |
| ResourceProviderMessage::UpdateState updateState{ |
| resourceProvider->info.id(), |
| update.resource_version_uuid(), |
| update.resources(), |
| std::move(operations)}; |
| |
| ResourceProviderMessage message; |
| message.type = ResourceProviderMessage::Type::UPDATE_STATE; |
| message.updateState = std::move(updateState); |
| |
| messages.put(std::move(message)); |
| } |
| |
| |
| void ResourceProviderManagerProcess::updatePublishResourcesStatus( |
| ResourceProvider* resourceProvider, |
| const Call::UpdatePublishResourcesStatus& update) |
| { |
| Try<id::UUID> uuid = id::UUID::fromBytes(update.uuid().value()); |
| |
| if (uuid.isError()) { |
| LOG(ERROR) |
| << "Ignoring UpdatePublishResourcesStatus from resource provider " |
| << resourceProvider->info.id() << ": " << uuid.error(); |
| |
| return; |
| } |
| |
| if (!resourceProvider->publishes.contains(uuid.get())) { |
| LOG(ERROR) |
| << "Ignoring UpdatePublishResourcesStatus from resource provider " |
| << resourceProvider->info.id() << ": Unknown UUID " << uuid.get(); |
| |
| return; |
| } |
| |
| LOG(INFO) |
| << "Received UPDATE_PUBLISH_RESOURCES_STATUS call for PUBLISH_RESOURCES" |
| << " event " << uuid.get() << " with " << update.status() |
| << " status from resource provider " << resourceProvider->info.id(); |
| |
| if (update.status() == Call::UpdatePublishResourcesStatus::OK) { |
| resourceProvider->publishes.at(uuid.get())->set(Nothing()); |
| } else { |
| // TODO(jieyu): Consider to include an error message in |
| // 'UpdatePublishResourcesStatus' and surface that to the caller. |
| resourceProvider->publishes.at(uuid.get())->fail( |
| "Received " + stringify(update.status()) + " status"); |
| } |
| |
| resourceProvider->publishes.erase(uuid.get()); |
| } |
| |
| |
| ResourceProviderID ResourceProviderManagerProcess::newResourceProviderId() |
| { |
| ResourceProviderID resourceProviderId; |
| resourceProviderId.set_value(id::UUID::random().toString()); |
| return resourceProviderId; |
| } |
| |
| |
| double ResourceProviderManagerProcess::gaugeSubscribed() |
| { |
| return static_cast<double>(resourceProviders.subscribed.size()); |
| } |
| |
| |
| ResourceProviderManagerProcess::Metrics::Metrics( |
| const ResourceProviderManagerProcess& manager) |
| : subscribed( |
| "resource_provider_manager/subscribed", |
| defer(manager, &ResourceProviderManagerProcess::gaugeSubscribed)), |
| subscriptions("resource_provider_manager/events/subscribe"), |
| disconnections("resource_provider_manager/events/disconnect") |
| { |
| process::metrics::add(subscribed); |
| process::metrics::add(subscriptions); |
| process::metrics::add(disconnections); |
| } |
| |
| |
| ResourceProviderManagerProcess::Metrics::~Metrics() |
| { |
| process::metrics::remove(subscribed); |
| process::metrics::remove(subscriptions); |
| process::metrics::remove(disconnections); |
| } |
| |
| |
| ResourceProviderManager::ResourceProviderManager(Owned<Registrar> registrar) |
| : process(new ResourceProviderManagerProcess(std::move(registrar))) |
| { |
| spawn(CHECK_NOTNULL(process.get())); |
| } |
| |
| |
| ResourceProviderManager::~ResourceProviderManager() |
| { |
| terminate(process.get()); |
| wait(process.get()); |
| } |
| |
| |
| Future<http::Response> ResourceProviderManager::api( |
| const http::Request& request, |
| const Option<Principal>& principal) const |
| { |
| return dispatch( |
| process.get(), |
| &ResourceProviderManagerProcess::api, |
| request, |
| principal); |
| } |
| |
| |
| void ResourceProviderManager::applyOperation( |
| const ApplyOperationMessage& message) const |
| { |
| return dispatch( |
| process.get(), |
| &ResourceProviderManagerProcess::applyOperation, |
| message); |
| } |
| |
| |
| void ResourceProviderManager::acknowledgeOperationStatus( |
| const AcknowledgeOperationStatusMessage& message) const |
| { |
| return dispatch( |
| process.get(), |
| &ResourceProviderManagerProcess::acknowledgeOperationStatus, |
| message); |
| } |
| |
| |
| void ResourceProviderManager::reconcileOperations( |
| const ReconcileOperationsMessage& message) const |
| { |
| return dispatch( |
| process.get(), |
| &ResourceProviderManagerProcess::reconcileOperations, |
| message); |
| } |
| |
| |
| Future<Nothing> ResourceProviderManager::removeResourceProvider( |
| const ResourceProviderID& resourceProviderId) const |
| { |
| return dispatch( |
| process.get(), |
| &ResourceProviderManagerProcess::removeResourceProvider, |
| resourceProviderId); |
| } |
| |
| Future<Nothing> ResourceProviderManager::publishResources( |
| const Resources& resources) |
| { |
| return dispatch( |
| process.get(), |
| &ResourceProviderManagerProcess::publishResources, |
| resources); |
| } |
| |
| |
| Queue<ResourceProviderMessage> ResourceProviderManager::messages() const |
| { |
| return process->messages; |
| } |
| |
| } // namespace internal { |
| } // namespace mesos { |
