You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by vi...@apache.org on 2015/08/13 23:13:36 UTC
[1/3] mesos git commit: Revert "Updated scheduler library to HTTP."
Repository: mesos
Updated Branches:
refs/heads/master dcdc79aa8 -> 49c114305
Revert "Updated scheduler library to HTTP."
This reverts commit 69704a28f115a23c4a8e1119bdda3d452674e679.
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/ddbd4297
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/ddbd4297
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/ddbd4297
Branch: refs/heads/master
Commit: ddbd429708f17caca593b4ba52cb10661066a39e
Parents: dcdc79a
Author: Vinod Kone <vi...@gmail.com>
Authored: Thu Aug 13 13:35:58 2015 -0700
Committer: Vinod Kone <vi...@gmail.com>
Committed: Thu Aug 13 13:35:58 2015 -0700
----------------------------------------------------------------------
src/common/http.hpp | 45 ---------
src/scheduler/scheduler.cpp | 203 +--------------------------------------
2 files changed, 5 insertions(+), 243 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/ddbd4297/src/common/http.hpp
----------------------------------------------------------------------
diff --git a/src/common/http.hpp b/src/common/http.hpp
index 0eab8ba..98a1270 100644
--- a/src/common/http.hpp
+++ b/src/common/http.hpp
@@ -25,7 +25,6 @@
#include <stout/hashmap.hpp>
#include <stout/json.hpp>
-#include <stout/protobuf.hpp>
namespace mesos {
@@ -48,21 +47,6 @@ enum class ContentType
};
-inline std::ostream& operator<<(std::ostream& stream, ContentType contentType)
-{
- switch (contentType) {
- case ContentType::PROTOBUF: {
- return stream << APPLICATION_PROTOBUF;
- }
- case ContentType::JSON: {
- return stream << APPLICATION_JSON;
- }
- }
-
- UNREACHABLE();
-}
-
-
// Serializes a protobuf message for transmission
// based on the HTTP content type.
std::string serialize(
@@ -70,35 +54,6 @@ std::string serialize(
const google::protobuf::Message& message);
-// Deserializes a string message into a protobuf message based on the
-// HTTP content type.
-template <typename Message>
-Try<Message> deserialize(
- ContentType contentType,
- const std::string& body)
-{
- switch (contentType) {
- case ContentType::PROTOBUF: {
- Message message;
- if (!message.ParseFromString(body)) {
- return Error("Failed to parse body into a protobuf object");
- }
- return message;
- }
- case ContentType::JSON: {
- Try<JSON::Value> value = JSON::parse(body);
- if (value.isError()) {
- return Error("Failed to parse body into JSON: " + value.error());
- }
-
- return ::protobuf::parse<Message>(value.get());
- }
- }
-
- UNREACHABLE();
-}
-
-
JSON::Object model(const Resources& resources);
JSON::Object model(const hashmap<std::string, Resources>& roleResources);
JSON::Object model(const Attributes& attributes);
http://git-wip-us.apache.org/repos/asf/mesos/blob/ddbd4297/src/scheduler/scheduler.cpp
----------------------------------------------------------------------
diff --git a/src/scheduler/scheduler.cpp b/src/scheduler/scheduler.cpp
index 37b5457..36d7052 100644
--- a/src/scheduler/scheduler.cpp
+++ b/src/scheduler/scheduler.cpp
@@ -26,7 +26,6 @@
#include <arpa/inet.h>
#include <iostream>
-#include <queue>
#include <string>
#include <sstream>
@@ -48,18 +47,13 @@
#include <stout/duration.hpp>
#include <stout/error.hpp>
#include <stout/flags.hpp>
-#include <stout/hashmap.hpp>
#include <stout/ip.hpp>
#include <stout/lambda.hpp>
#include <stout/nothing.hpp>
#include <stout/option.hpp>
#include <stout/os.hpp>
-#include <stout/recordio.hpp>
#include <stout/uuid.hpp>
-#include "common/http.hpp"
-#include "common/recordio.hpp"
-
#include "internal/devolve.hpp"
#include "internal/evolve.hpp"
@@ -83,17 +77,8 @@ using std::queue;
using std::string;
using std::vector;
-using mesos::internal::recordio::Reader;
-
-using process::Owned;
using process::wait; // Necessary on some OS's to disambiguate.
-using process::http::Pipe;
-using process::http::post;
-using process::http::Response;
-
-using ::recordio::Decoder;
-
namespace mesos {
namespace v1 {
namespace scheduler {
@@ -106,12 +91,10 @@ class MesosProcess : public ProtobufProcess<MesosProcess>
public:
MesosProcess(
const string& master,
- ContentType _contentType,
const lambda::function<void(void)>& _connected,
const lambda::function<void(void)>& _disconnected,
lambda::function<void(const queue<Event>&)> _received)
: ProcessBase(ID::generate("scheduler")),
- contentType(_contentType),
connected(_connected),
disconnected(_disconnected),
received(_received),
@@ -177,8 +160,6 @@ public:
virtual ~MesosProcess()
{
- disconnect();
-
// Check and see if we need to shutdown a local cluster.
if (local) {
local::shutdown();
@@ -191,7 +172,7 @@ public:
// TODO(benh): Move this to 'protected'.
using ProtobufProcess<MesosProcess>::send;
- void send(const Call& call)
+ void send(Call call)
{
if (master.isNone()) {
drop(call, "Disconnected");
@@ -208,34 +189,7 @@ public:
// TODO(vinod): Add support for sending MESSAGE calls directly
// to the slave, instead of relaying it through the master, as
// the scheduler driver does.
-
- const string body = serialize(contentType, call);
- const hashmap<string, string> headers{{"Accept", stringify(contentType)}};
-
- Future<Response> response;
-
- if (call.type() == Call::SUBSCRIBE) {
- // Each subscription requires a new connection.
- disconnect();
-
- // Send a streaming request for Subscribe call.
- response = process::http::streaming::post(
- master.get(),
- "api/v1/scheduler",
- headers,
- body,
- stringify(contentType));
- } else {
- response = post(
- master.get(),
- "api/v1/scheduler",
- headers,
- body,
- stringify(contentType));
- }
-
- response
- .onAny(defer(self(), &Self::_send, call, lambda::_1));
+ send(master.get(), devolve(call));
}
protected:
@@ -255,9 +209,6 @@ protected:
return;
}
- // Disconnect the reader upon a master detection callback.
- disconnect();
-
if (future.get().isNone()) {
master = None();
@@ -309,7 +260,7 @@ protected:
error->set_message(message);
- receive(event, true);
+ receive(None(), event);
}
void drop(const Call& call, const string& message)
@@ -317,145 +268,7 @@ protected:
LOG(WARNING) << "Dropping " << call.type() << ": " << message;
}
- void _send(const Call& call, const Future<Response>& response)
- {
- CHECK(!response.isDiscarded());
-
- // This can happen during a master failover or a network blip
- // causing the socket to timeout. Eventually, the scheduler would
- // detect the disconnection via ZK(disconnect()) or lack of heartbeats.
- if (response.isFailed()) {
- LOG(ERROR) << "Request for call type " << call.type() << " failed: "
- << response.failure();
- return;
- }
-
- if (call.type() == Call::SUBSCRIBE &&
- response.get().status == process::http::statuses[200] ) {
- CHECK_EQ(response.get().type, http::Response::PIPE);
- CHECK_SOME(response.get().reader);
-
- Pipe::Reader reader = response.get().reader.get();
-
- auto deserializer =
- lambda::bind(deserialize<Event>, contentType, lambda::_1);
-
- Owned<Reader<Event>> decoder(
- new Reader<Event>(Decoder<Event>(deserializer), reader));
-
- connection = Connection {reader, decoder};
-
- read();
- return;
- }
-
- if (call.type() != Call::SUBSCRIBE &&
- response.get().status == process::http::statuses[202] ) {
- return;
- }
-
- // We should be able to get here only for AuthN errors which is not
- // yet supported for HTTP frameworks. The other possible scenario
- // can be that the master was not able to de-serialize the Call
- // message. Since we validate the Call messages before sending, this
- // can only happen when a packet corruption happens.
- error("Received unexpected '" + response.get().status + "' for " +
- stringify(call.type()) + " call: " + response.get().body);
- }
-
- void read()
- {
- connection.get().decoder->read()
- .onAny(defer(self(),
- &Self::_read,
- connection.get().reader,
- lambda::_1));
- }
-
- void _read(const Pipe::Reader& reader, const Future<Result<Event>>& event)
- {
- CHECK(!event.isDiscarded());
-
- // Ignore enqueued events from the previous Subscribe call reader.
- if (!connection.isSome() || connection.get().reader != reader) {
- VLOG(1) << "Ignoring event from old stale connection";
- return;
- }
-
- // This could happen if the master failed over while sending a response.
- // It's fine to drop this as the scheduler would detect the
- // disconnection via ZK(disconnect) or lack of heartbeats.
- if (event.isFailed()) {
- LOG(ERROR) << "Failed to decode the stream of events: "
- << event.failure();
- return;
- }
-
- if (!event.get().isSome()) {
- // It's fine to drop this as the scheduler would detect the
- // disconnection via ZK(disconnect) or lack of heartbeats.
- LOG(ERROR) << "End-Of-File received from master."
- << " The master closed the event stream";
- return;
- }
-
- if (event.get().isError()) {
- error("Failed to de-serialize event: " + event.get().error());
- } else {
- receive(event.get().get(), false);
- }
-
- read();
- }
-
- void receive(const Event& event, bool isLocallyInjected)
- {
- // Check if we're disconnected but received an event.
- if (!isLocallyInjected && master.isNone()) {
- LOG(WARNING) << "Ignoring " << stringify(event.type())
- << " event because we're disconnected";
- return;
- }
-
- if (isLocallyInjected) {
- VLOG(1) << "Enqueuing locally injected event "
- << stringify(event.type());
- }
-
- // Queue up the event and invoke the 'received' callback if this
- // is the first event (between now and when the 'received'
- // callback actually gets invoked more events might get queued).
- events.push(event);
-
- if (events.size() == 1) {
- mutex.lock()
- .then(defer(self(), &Self::_receive))
- .onAny(lambda::bind(&Mutex::unlock, mutex));
- }
- }
-
- void disconnect()
- {
- if (connection.isSome()) {
- if (!connection.get().reader.close()) {
- LOG(WARNING) << "HTTP connection was already closed";
- }
- }
-
- connection = None();
- }
-
private:
- struct Connection
- {
- Pipe::Reader reader;
- process::Owned<Reader<Event>> decoder;
- };
-
- Option<Connection> connection;
-
- ContentType contentType;
-
Mutex mutex; // Used to serialize the callback invocations.
lambda::function<void(void)> connected;
@@ -478,14 +291,8 @@ Mesos::Mesos(
const lambda::function<void(void)>& disconnected,
const lambda::function<void(const queue<Event>&)>& received)
{
- // TODO(anand): Make ContentType as a constructor argument.
- process = new MesosProcess(
- master,
- ContentType::PROTOBUF,
- connected,
- disconnected,
- received);
-
+ process =
+ new MesosProcess(master, connected, disconnected, received);
spawn(process);
}
[2/3] mesos git commit: Revert "Deleted old style message handling
from the scheduler library."
Posted by vi...@apache.org.
Revert "Deleted old style message handling from the scheduler library."
This reverts commit 138ca6903a3adfcf21783ae613e120b785245304.
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/2926208c
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/2926208c
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/2926208c
Branch: refs/heads/master
Commit: 2926208c4108a8467cba00d6d49b549e7286f5a1
Parents: ddbd429
Author: Vinod Kone <vi...@gmail.com>
Authored: Thu Aug 13 13:36:09 2015 -0700
Committer: Vinod Kone <vi...@gmail.com>
Committed: Thu Aug 13 13:36:09 2015 -0700
----------------------------------------------------------------------
src/scheduler/scheduler.cpp | 89 ++++++++++++++++++++++++++++++++++++++++
1 file changed, 89 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/2926208c/src/scheduler/scheduler.cpp
----------------------------------------------------------------------
diff --git a/src/scheduler/scheduler.cpp b/src/scheduler/scheduler.cpp
index 36d7052..3fbe383 100644
--- a/src/scheduler/scheduler.cpp
+++ b/src/scheduler/scheduler.cpp
@@ -195,6 +195,16 @@ public:
protected:
virtual void initialize()
{
+ install<FrameworkRegisteredMessage>(&MesosProcess::receive);
+ install<FrameworkReregisteredMessage>(&MesosProcess::receive);
+ install<ResourceOffersMessage>(&MesosProcess::receive);
+ install<RescindResourceOfferMessage>(&MesosProcess::receive);
+ install<StatusUpdateMessage>(&MesosProcess::receive);
+ install<LostSlaveMessage>(&MesosProcess::receive);
+ install<ExitedExecutorMessage>(&MesosProcess::receive);
+ install<ExecutorToFrameworkMessage>(&MesosProcess::receive);
+ install<FrameworkErrorMessage>(&MesosProcess::receive);
+
// Start detecting masters.
detector->detect()
.onAny(defer(self(), &MesosProcess::detected, lambda::_1));
@@ -242,6 +252,40 @@ protected:
return async(connected);
}
+ // NOTE: A None 'from' is possible when an event is injected locally.
+ void receive(const Option<UPID>& from, const Event& event)
+ {
+ // Check if we're disconnected but received an event.
+ if (from.isSome() && master.isNone()) {
+ VLOG(1) << "Ignoring " << stringify(event.type())
+ << " event because we're disconnected";
+ return;
+ } else if (from.isSome() && master != from) {
+ VLOG(1)
+ << "Ignoring " << stringify(event.type())
+ << " event because it was sent from '" << from.get()
+ << "' instead of the leading master '" << master.get() << "'";
+ return;
+ }
+
+ // Note that if 'from' is None we're locally injecting this event
+ // so we always want to enqueue it even if we're not connected!
+
+ VLOG(1) << "Enqueuing event " << stringify(event.type()) << " from "
+ << (from.isNone() ? "(locally injected)" : from.get());
+
+ // Queue up the event and invoke the 'received' callback if this
+ // is the first event (between now and when the 'received'
+ // callback actually gets invoked more events might get queued).
+ events.push(event);
+
+ if (events.size() == 1) {
+ mutex.lock()
+ .then(defer(self(), &Self::_receive))
+ .onAny(lambda::bind(&Mutex::unlock, mutex));
+ }
+ }
+
Future<Nothing> _receive()
{
Future<Nothing> future = async(received, events);
@@ -249,6 +293,51 @@ protected:
return future;
}
+ void receive(const UPID& from, const FrameworkRegisteredMessage& message)
+ {
+ receive(from, evolve(message));
+ }
+
+ void receive(const UPID& from, const FrameworkReregisteredMessage& message)
+ {
+ receive(from, evolve(message));
+ }
+
+ void receive(const UPID& from, const ResourceOffersMessage& message)
+ {
+ receive(from, evolve(message));
+ }
+
+ void receive(const UPID& from, const RescindResourceOfferMessage& message)
+ {
+ receive(from, evolve(message));
+ }
+
+ void receive(const UPID& from, const StatusUpdateMessage& message)
+ {
+ receive(from, evolve(message));
+ }
+
+ void receive(const UPID& from, const LostSlaveMessage& message)
+ {
+ receive(from, evolve(message));
+ }
+
+ void receive(const UPID& from, const ExitedExecutorMessage& message)
+ {
+ receive(from, evolve(message));
+ }
+
+ void receive(const UPID& from, const ExecutorToFrameworkMessage& message)
+ {
+ receive(from, evolve(message));
+ }
+
+ void receive(const UPID& from, const FrameworkErrorMessage& message)
+ {
+ receive(from, evolve(message));
+ }
+
// Helper for injecting an ERROR event.
void error(const string& message)
{
[3/3] mesos git commit: Added output stream operator for ContentType
enum.
Posted by vi...@apache.org.
Added output stream operator for ContentType enum.
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/49c11430
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/49c11430
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/49c11430
Branch: refs/heads/master
Commit: 49c114305cc81ed78b326c23427fb6e985a8c3d7
Parents: 2926208
Author: Vinod Kone <vi...@gmail.com>
Authored: Thu Aug 13 14:13:05 2015 -0700
Committer: Vinod Kone <vi...@gmail.com>
Committed: Thu Aug 13 14:13:05 2015 -0700
----------------------------------------------------------------------
src/common/http.hpp | 15 +++++++++++++++
1 file changed, 15 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/49c11430/src/common/http.hpp
----------------------------------------------------------------------
diff --git a/src/common/http.hpp b/src/common/http.hpp
index 98a1270..6f4f686 100644
--- a/src/common/http.hpp
+++ b/src/common/http.hpp
@@ -47,6 +47,21 @@ enum class ContentType
};
+inline std::ostream& operator<<(std::ostream& stream, ContentType contentType)
+{
+ switch (contentType) {
+ case ContentType::PROTOBUF: {
+ return stream << APPLICATION_PROTOBUF;
+ }
+ case ContentType::JSON: {
+ return stream << APPLICATION_JSON;
+ }
+ }
+
+ UNREACHABLE();
+}
+
+
// Serializes a protobuf message for transmission
// based on the HTTP content type.
std::string serialize(