You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by gr...@apache.org on 2018/12/14 21:32:06 UTC
[mesos] 02/06: Refactored master and agent streaming connections.
This is an automated email from the ASF dual-hosted git repository.
grag pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit ba46deb2ba31bd7f3d9bff3db979a1e850eedf0c
Author: Joseph Wu <jo...@mesosphere.io>
AuthorDate: Thu Dec 13 16:34:24 2018 -0800
Refactored master and agent streaming connections.
This moves the very similar `HttpConnection` classes inside the
master and agent into a common header. The refactored
`StreamingHttpConnection<Event>` is more explicitly named to avoid
potentially clashing with the libprocess HTTP helpers.
This also moves the master's heartbeater helper into a new header
and transforms it into an RAII libprocess actor wrapper. The
heartbeater depends on this `StreamingHttpConnection` and is currently
used by the master for heartbeating the operator event stream
and HTTP framework connection. A later patch will use this heartbeater
for agent->executor heartbeats.
Review: https://reviews.apache.org/r/69472/
---
src/Makefile.am | 1 +
src/common/heartbeater.hpp | 138 ++++++++++++++++++++++++++++++++++++
src/common/http.hpp | 45 ++++++++++++
src/master/framework.cpp | 37 ++++------
src/master/http.cpp | 11 +--
src/master/master.cpp | 35 ++++++----
src/master/master.hpp | 169 +++++++++------------------------------------
src/slave/http.cpp | 4 +-
src/slave/slave.cpp | 2 +-
src/slave/slave.hpp | 39 +----------
10 files changed, 264 insertions(+), 217 deletions(-)
diff --git a/src/Makefile.am b/src/Makefile.am
index 3550051..7a4904a 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -1011,6 +1011,7 @@ libmesos_no_3rdparty_la_SOURCES += \
common/build.hpp \
common/command_utils.cpp \
common/command_utils.hpp \
+ common/heartbeater.hpp \
common/http.cpp \
common/http.hpp \
common/parse.hpp \
diff --git a/src/common/heartbeater.hpp b/src/common/heartbeater.hpp
new file mode 100644
index 0000000..f7c3a80
--- /dev/null
+++ b/src/common/heartbeater.hpp
@@ -0,0 +1,138 @@
+// 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 __COMMON_HEARTBEATER_HPP__
+#define __COMMON_HEARTBEATER_HPP__
+
+#include <string>
+
+#include <process/delay.hpp>
+#include <process/http.hpp>
+#include <process/process.hpp>
+#include <process/owned.hpp>
+
+#include <stout/duration.hpp>
+#include <stout/lambda.hpp>
+#include <stout/option.hpp>
+
+#include "common/http.hpp"
+
+namespace mesos {
+namespace internal {
+
+// This process periodically sends heartbeats to a given HTTP streaming
+// response. The optional delay parameter is used to specify the delay
+// period before sending the first heartbeat. The optional callback parameter
+// will be called each time a heartbeat is sent.
+template<typename Message, typename Event>
+class ResponseHeartbeaterProcess
+ : public process::Process<ResponseHeartbeaterProcess<Message, Event>>
+{
+public:
+ ResponseHeartbeaterProcess(
+ const std::string& _logMessage,
+ const Message& _heartbeatMessage,
+ const StreamingHttpConnection<Event>& _connection,
+ const Duration& _interval,
+ const Option<Duration>& _delay = None(),
+ const Option<lambda::function<void()>>& _callback = None())
+ : process::ProcessBase(process::ID::generate("heartbeater")),
+ logMessage(_logMessage),
+ heartbeatMessage(_heartbeatMessage),
+ connection(_connection),
+ interval(_interval),
+ delay(_delay),
+ callback(_callback) {}
+
+protected:
+ void initialize() override
+ {
+ if (delay.isSome()) {
+ process::delay(
+ delay.get(),
+ this,
+ &ResponseHeartbeaterProcess::heartbeat);
+ } else {
+ heartbeat();
+ }
+ }
+
+private:
+ void heartbeat()
+ {
+ // Only send a heartbeat if the connection is not closed.
+ if (connection.closed().isPending()) {
+ VLOG(2) << "Sending heartbeat to " << logMessage;
+
+ if (callback.isSome()) {
+ callback.get()();
+ }
+
+ connection.send(heartbeatMessage);
+ }
+
+ process::delay(interval, this, &ResponseHeartbeaterProcess::heartbeat);
+ }
+
+ const std::string logMessage;
+ const Message heartbeatMessage;
+ StreamingHttpConnection<Event> connection;
+ const Duration interval;
+ const Option<Duration> delay;
+ const Option<lambda::function<void()>> callback;
+};
+
+
+template<typename Message, typename Event>
+class ResponseHeartbeater
+{
+public:
+ ResponseHeartbeater(
+ const std::string& _logMessage,
+ const Message& _heartbeatMessage,
+ const StreamingHttpConnection<Event>& _connection,
+ const Duration& _interval,
+ const Option<Duration>& _delay = None(),
+ const Option<lambda::function<void()>>& _callback = None())
+ : process(new ResponseHeartbeaterProcess<Message, Event>(
+ _logMessage,
+ _heartbeatMessage,
+ _connection,
+ _interval,
+ _delay,
+ _callback))
+ {
+ process::spawn(process.get());
+ }
+
+ ~ResponseHeartbeater()
+ {
+ process::terminate(process.get());
+ process::wait(process.get());
+ }
+
+ // Not copyable, not assignable.
+ ResponseHeartbeater(const ResponseHeartbeater&) = delete;
+ ResponseHeartbeater& operator=(const ResponseHeartbeater&) = delete;
+
+private:
+ const process::Owned<ResponseHeartbeaterProcess<Message, Event>> process;
+};
+
+} // namespace internal {
+} // namespace mesos {
+
+#endif // __COMMON_HEARTBEATER_HPP__
diff --git a/src/common/http.hpp b/src/common/http.hpp
index ac9ed5e..a11ce3c 100644
--- a/src/common/http.hpp
+++ b/src/common/http.hpp
@@ -36,7 +36,11 @@
#include <stout/json.hpp>
#include <stout/jsonify.hpp>
#include <stout/protobuf.hpp>
+#include <stout/recordio.hpp>
#include <stout/unreachable.hpp>
+#include <stout/uuid.hpp>
+
+#include "internal/evolve.hpp"
// TODO(benh): Remove this once we get C++14 as an enum should have a
// default hash.
@@ -138,6 +142,47 @@ Try<Message> deserialize(
bool streamingMediaType(ContentType contentType);
+// Represents the streaming HTTP connection to a client, such as a framework,
+// executor, or operator subscribed to the '/api/vX' endpoint.
+// The `Event` template is the evolved message being sent to the client,
+// e.g. `v1::scheduler::Event`, `v1::master::Event`, or `v1::executor::Event`.
+template <typename Event>
+struct StreamingHttpConnection
+{
+ StreamingHttpConnection(
+ const process::http::Pipe::Writer& _writer,
+ ContentType _contentType,
+ id::UUID _streamId = id::UUID::random())
+ : writer(_writer),
+ contentType(_contentType),
+ encoder(lambda::bind(serialize, contentType, lambda::_1)),
+ streamId(_streamId) {}
+
+ // Converts the message to the templated `Event`, via `evolve()`,
+ // before sending.
+ template <typename Message>
+ bool send(const Message& message)
+ {
+ return writer.write(encoder.encode(evolve(message)));
+ }
+
+ bool close()
+ {
+ return writer.close();
+ }
+
+ process::Future<Nothing> closed() const
+ {
+ return writer.readerClosed();
+ }
+
+ process::http::Pipe::Writer writer;
+ ContentType contentType;
+ ::recordio::Encoder<Event> encoder;
+ id::UUID streamId;
+};
+
+
JSON::Object model(const Resources& resources);
JSON::Object model(const hashmap<std::string, Resources>& roleResources);
JSON::Object model(const Attributes& attributes);
diff --git a/src/master/framework.cpp b/src/master/framework.cpp
index 36eda9f..4089cf4 100644
--- a/src/master/framework.cpp
+++ b/src/master/framework.cpp
@@ -16,6 +16,7 @@
#include "master/master.hpp"
+#include "common/heartbeater.hpp"
#include "common/protobuf_utils.hpp"
namespace mesos {
@@ -38,7 +39,7 @@ Framework::Framework(
Master* const master,
const Flags& masterFlags,
const FrameworkInfo& info,
- const HttpConnection& _http,
+ const StreamingHttpConnection<v1::scheduler::Event>& _http,
const process::Time& time)
: Framework(master, masterFlags, info, ACTIVE, time)
{
@@ -624,7 +625,8 @@ void Framework::updateConnection(const process::UPID& newPid)
}
-void Framework::updateConnection(const HttpConnection& newHttp)
+void Framework::updateConnection(
+ const StreamingHttpConnection<v1::scheduler::Event>& newHttp)
{
if (pid.isSome()) {
// Wipe the PID if this is an upgrade from PID to HTTP.
@@ -653,19 +655,12 @@ void Framework::closeHttpConnection()
}
http = None();
-
- CHECK_SOME(heartbeater);
-
- terminate(heartbeater->get());
- wait(heartbeater->get());
-
- heartbeater = None();
+ heartbeater.reset();
}
void Framework::heartbeat()
{
- CHECK_NONE(heartbeater);
CHECK_SOME(http);
// TODO(vinod): Make heartbeat interval configurable and include
@@ -673,18 +668,16 @@ void Framework::heartbeat()
scheduler::Event event;
event.set_type(scheduler::Event::HEARTBEAT);
- heartbeater =
- new Heartbeater<scheduler::Event, v1::scheduler::Event>(
- "framework " + stringify(info.id()),
- event,
- http.get(),
- DEFAULT_HEARTBEAT_INTERVAL,
- None(),
- [this](const scheduler::Event& event) {
- this->metrics.incrementEvent(event);
- });
-
- process::spawn(heartbeater->get());
+ heartbeater.reset(
+ new ResponseHeartbeater<scheduler::Event, v1::scheduler::Event>(
+ "framework " + stringify(info.id()),
+ event,
+ http.get(),
+ DEFAULT_HEARTBEAT_INTERVAL,
+ None(),
+ [this, event]() {
+ this->metrics.incrementEvent(event);
+ }));
}
diff --git a/src/master/http.cpp b/src/master/http.cpp
index 68ee2a6..e38513b 100644
--- a/src/master/http.cpp
+++ b/src/master/http.cpp
@@ -412,7 +412,8 @@ Future<Response> Master::Http::subscribe(
ok.type = Response::PIPE;
ok.reader = pipe.reader();
- HttpConnection http{pipe.writer(), contentType, id::UUID::random()};
+ StreamingHttpConnection<v1::master::Event> http(
+ pipe.writer(), contentType);
mesos::master::Event event;
event.set_type(mesos::master::Event::SUBSCRIBED);
@@ -422,11 +423,11 @@ Future<Response> Master::Http::subscribe(
event.mutable_subscribed()->set_heartbeat_interval_seconds(
DEFAULT_HEARTBEAT_INTERVAL.secs());
- http.send<mesos::master::Event, v1::master::Event>(event);
+ http.send(event);
mesos::master::Event heartbeatEvent;
heartbeatEvent.set_type(mesos::master::Event::HEARTBEAT);
- http.send<mesos::master::Event, v1::master::Event>(heartbeatEvent);
+ http.send(heartbeatEvent);
// Master::subscribe will start the heartbeater process, which should
// only happen after `SUBSCRIBED` event is sent.
@@ -598,7 +599,9 @@ Future<Response> Master::Http::scheduler(
id::UUID streamId = id::UUID::random();
ok.headers["Mesos-Stream-Id"] = streamId.toString();
- HttpConnection http {pipe.writer(), acceptType, streamId};
+ StreamingHttpConnection<v1::scheduler::Event> http(
+ pipe.writer(), acceptType, streamId);
+
master->subscribe(http, call.subscribe());
return ok;
diff --git a/src/master/master.cpp b/src/master/master.cpp
index 3de0fd3..5443bd1 100644
--- a/src/master/master.cpp
+++ b/src/master/master.cpp
@@ -1235,7 +1235,9 @@ void Master::finalize()
}
-void Master::exited(const FrameworkID& frameworkId, const HttpConnection& http)
+void Master::exited(
+ const FrameworkID& frameworkId,
+ const StreamingHttpConnection<v1::scheduler::Event>& http)
{
foreachvalue (Framework* framework, frameworks.registered) {
if (framework->http.isSome() && framework->http->writer == http.writer) {
@@ -2496,7 +2498,7 @@ void Master::reregisterFramework(
void Master::subscribe(
- HttpConnection http,
+ StreamingHttpConnection<v1::scheduler::Event> http,
const scheduler::Call::Subscribe& subscribe)
{
// TODO(anand): Authenticate the framework.
@@ -2596,7 +2598,7 @@ void Master::subscribe(
// Need to disambiguate for the compiler.
void (Master::*_subscribe)(
- HttpConnection,
+ StreamingHttpConnection<v1::scheduler::Event>,
const FrameworkInfo&,
bool,
const set<string>&,
@@ -2614,7 +2616,7 @@ void Master::subscribe(
void Master::_subscribe(
- HttpConnection http,
+ StreamingHttpConnection<v1::scheduler::Event> http,
const FrameworkInfo& frameworkInfo,
bool force,
const set<string>& suppressedRoles,
@@ -9927,7 +9929,8 @@ void Master::addFramework(
} else {
CHECK_SOME(framework->http);
- const HttpConnection& http = framework->http.get();
+ const StreamingHttpConnection<v1::scheduler::Event>& http =
+ framework->http.get();
http.closed()
.onAny(defer(self(), &Self::exited, framework->id(), http));
@@ -10017,7 +10020,7 @@ Try<Nothing> Master::activateRecoveredFramework(
Framework* framework,
const FrameworkInfo& frameworkInfo,
const Option<UPID>& pid,
- const Option<HttpConnection>& http,
+ const Option<StreamingHttpConnection<v1::scheduler::Event>>& http,
const set<string>& suppressedRoles)
{
// Exactly one of `pid` or `http` must be provided.
@@ -10095,7 +10098,9 @@ Try<Nothing> Master::activateRecoveredFramework(
}
-void Master::failoverFramework(Framework* framework, const HttpConnection& http)
+void Master::failoverFramework(
+ Framework* framework,
+ const StreamingHttpConnection<v1::scheduler::Event>& http)
{
CHECK_NOTNULL(framework);
@@ -12011,7 +12016,7 @@ void Master::Subscribers::Subscriber::send(
if (approvers->approved<VIEW_TASK>(
event->task_added().task(), *frameworkInfo) &&
approvers->approved<VIEW_FRAMEWORK>(*frameworkInfo)) {
- http.send<mesos::master::Event, v1::master::Event>(*event);
+ http.send(*event);
}
break;
}
@@ -12021,7 +12026,7 @@ void Master::Subscribers::Subscriber::send(
if (approvers->approved<VIEW_TASK>(*task, *frameworkInfo) &&
approvers->approved<VIEW_FRAMEWORK>(*frameworkInfo)) {
- http.send<mesos::master::Event, v1::master::Event>(*event);
+ http.send(*event);
}
break;
}
@@ -12052,7 +12057,7 @@ void Master::Subscribers::Subscriber::send(
}
}
- http.send<mesos::master::Event, v1::master::Event>(event_);
+ http.send(event_);
}
break;
}
@@ -12083,14 +12088,14 @@ void Master::Subscribers::Subscriber::send(
}
}
- http.send<mesos::master::Event, v1::master::Event>(event_);
+ http.send(event_);
}
break;
}
case mesos::master::Event::FRAMEWORK_REMOVED: {
if (approvers->approved<VIEW_FRAMEWORK>(
event->framework_removed().framework_info())) {
- http.send<mesos::master::Event, v1::master::Event>(*event);
+ http.send(*event);
}
break;
}
@@ -12108,14 +12113,14 @@ void Master::Subscribers::Subscriber::send(
}
}
- http.send<mesos::master::Event, v1::master::Event>(event_);
+ http.send(event_);
break;
}
case mesos::master::Event::AGENT_REMOVED:
case mesos::master::Event::SUBSCRIBED:
case mesos::master::Event::HEARTBEAT:
case mesos::master::Event::UNKNOWN:
- http.send<mesos::master::Event, v1::master::Event>(*event);
+ http.send(*event);
break;
}
}
@@ -12136,7 +12141,7 @@ void Master::exited(const id::UUID& id)
void Master::subscribe(
- const HttpConnection& http,
+ const StreamingHttpConnection<v1::master::Event>& http,
const Option<Principal>& principal)
{
LOG(INFO) << "Added subscriber " << http.streamId
diff --git a/src/master/master.hpp b/src/master/master.hpp
index c7becfa..0c1821c 100644
--- a/src/master/master.hpp
+++ b/src/master/master.hpp
@@ -67,6 +67,7 @@
#include <stout/try.hpp>
#include <stout/uuid.hpp>
+#include "common/heartbeater.hpp"
#include "common/http.hpp"
#include "common/resources_utils.hpp"
@@ -328,109 +329,6 @@ inline std::ostream& operator<<(std::ostream& stream, const Slave& slave)
}
-// Represents the streaming HTTP connection to a framework or a client
-// subscribed to the '/api/vX' endpoint.
-struct HttpConnection
-{
- HttpConnection(const process::http::Pipe::Writer& _writer,
- ContentType _contentType,
- id::UUID _streamId)
- : writer(_writer),
- contentType(_contentType),
- streamId(_streamId) {}
-
- // We need to evolve the internal old style message/unversioned event into a
- // versioned event e.g., `v1::scheduler::Event` or `v1::master::Event`.
- template <typename Message, typename Event = v1::scheduler::Event>
- bool send(const Message& message)
- {
- ::recordio::Encoder<Event> encoder (lambda::bind(
- serialize, contentType, lambda::_1));
-
- return writer.write(encoder.encode(evolve(message)));
- }
-
- bool close()
- {
- return writer.close();
- }
-
- process::Future<Nothing> closed() const
- {
- return writer.readerClosed();
- }
-
- process::http::Pipe::Writer writer;
- ContentType contentType;
- id::UUID streamId;
-};
-
-
-// This process periodically sends heartbeats to a given HTTP connection.
-// The `Message` template parameter is the type of the heartbeat event passed
-// into the heartbeater during construction, while the `Event` template
-// parameter is the versioned event type which is sent to the client.
-// The optional delay parameter is used to specify the delay period before it
-// sends the first heartbeat.
-template <typename Message, typename Event>
-class Heartbeater : public process::Process<Heartbeater<Message, Event>>
-{
-public:
- Heartbeater(const std::string& _logMessage,
- const Message& _heartbeatMessage,
- const HttpConnection& _http,
- const Duration& _interval,
- const Option<Duration>& _delay = None(),
- const Option<lambda::function<void(const Message&)>>&
- _callback = None())
- : process::ProcessBase(process::ID::generate("heartbeater")),
- logMessage(_logMessage),
- heartbeatMessage(_heartbeatMessage),
- http(_http),
- interval(_interval),
- delay(_delay),
- callback(_callback) {}
-
-protected:
- void initialize() override
- {
- if (delay.isSome()) {
- process::delay(
- delay.get(),
- this,
- &Heartbeater<Message, Event>::heartbeat);
- } else {
- heartbeat();
- }
- }
-
-private:
- void heartbeat()
- {
- // Only send a heartbeat if the connection is not closed.
- if (http.closed().isPending()) {
- VLOG(2) << "Sending heartbeat to " << logMessage;
-
- if (callback.isSome()) {
- callback.get()(heartbeatMessage);
- }
-
- Message message(heartbeatMessage);
- http.send<Message, Event>(message);
- }
-
- process::delay(interval, this, &Heartbeater<Message, Event>::heartbeat);
- }
-
- const std::string logMessage;
- const Message heartbeatMessage;
- HttpConnection http;
- const Duration interval;
- const Option<Duration> delay;
- const Option<lambda::function<void(const Message&)>> callback;
-};
-
-
class Master : public ProtobufProcess<Master>
{
public:
@@ -596,7 +494,10 @@ protected:
void consume(process::ExitedEvent&& event) override;
void exited(const process::UPID& pid) override;
- void exited(const FrameworkID& frameworkId, const HttpConnection& http);
+ void exited(
+ const FrameworkID& frameworkId,
+ const StreamingHttpConnection<v1::scheduler::Event>& http);
+
void _exited(Framework* framework);
// Invoked upon noticing a subscriber disconnection.
@@ -703,7 +604,7 @@ protected:
Framework* framework,
const FrameworkInfo& frameworkInfo,
const Option<process::UPID>& pid,
- const Option<HttpConnection>& http,
+ const Option<StreamingHttpConnection<v1::scheduler::Event>>& http,
const std::set<std::string>& suppressedRoles);
// Replace the scheduler for a framework with a new process ID, in
@@ -712,7 +613,9 @@ protected:
// Replace the scheduler for a framework with a new HTTP connection,
// in the event of a scheduler failover.
- void failoverFramework(Framework* framework, const HttpConnection& http);
+ void failoverFramework(
+ Framework* framework,
+ const StreamingHttpConnection<v1::scheduler::Event>& http);
void _failoverFramework(Framework* framework);
@@ -1073,11 +976,11 @@ private:
mesos::scheduler::Call&& call);
void subscribe(
- HttpConnection http,
+ StreamingHttpConnection<v1::scheduler::Event> http,
const mesos::scheduler::Call::Subscribe& subscribe);
void _subscribe(
- HttpConnection http,
+ StreamingHttpConnection<v1::scheduler::Event> http,
const FrameworkInfo& frameworkInfo,
bool force,
const std::set<std::string>& suppressedRoles,
@@ -1096,7 +999,7 @@ private:
// Subscribes a client to the 'api/vX' endpoint.
void subscribe(
- const HttpConnection& http,
+ const StreamingHttpConnection<v1::master::Event>& http,
const Option<process::http::authentication::Principal>& principal);
void teardown(Framework* framework);
@@ -2158,25 +2061,20 @@ private:
struct Subscriber
{
Subscriber(
- const HttpConnection& _http,
+ const StreamingHttpConnection<v1::master::Event>& _http,
const Option<process::http::authentication::Principal> _principal)
: http(_http),
- principal(_principal)
- {
- mesos::master::Event event;
- event.set_type(mesos::master::Event::HEARTBEAT);
-
- heartbeater =
- process::Owned<Heartbeater<mesos::master::Event, v1::master::Event>>(
- new Heartbeater<mesos::master::Event, v1::master::Event>(
- "subscriber " + stringify(http.streamId),
- event,
- http,
- DEFAULT_HEARTBEAT_INTERVAL,
- DEFAULT_HEARTBEAT_INTERVAL));
-
- process::spawn(heartbeater.get());
- }
+ heartbeater(
+ "subscriber " + stringify(http.streamId),
+ []() {
+ mesos::master::Event event;
+ event.set_type(mesos::master::Event::HEARTBEAT);
+ return event;
+ }(),
+ http,
+ DEFAULT_HEARTBEAT_INTERVAL,
+ DEFAULT_HEARTBEAT_INTERVAL),
+ principal(_principal) {}
// Not copyable, not assignable.
Subscriber(const Subscriber&) = delete;
@@ -2197,14 +2095,10 @@ private:
// after passing ownership to the `Subscriber` object. See MESOS-5843
// for more details.
http.close();
-
- terminate(heartbeater.get());
- wait(heartbeater.get());
}
- HttpConnection http;
- process::Owned<Heartbeater<mesos::master::Event, v1::master::Event>>
- heartbeater;
+ StreamingHttpConnection<v1::master::Event> http;
+ ResponseHeartbeater<mesos::master::Event, v1::master::Event> heartbeater;
const Option<process::http::authentication::Principal> principal;
};
@@ -2387,7 +2281,7 @@ struct Framework
Framework(Master* const master,
const Flags& masterFlags,
const FrameworkInfo& info,
- const HttpConnection& _http,
+ const StreamingHttpConnection<v1::scheduler::Event>& _http,
const process::Time& time = process::Clock::now());
Framework(Master* const master,
@@ -2456,7 +2350,8 @@ struct Framework
void updateConnection(const process::UPID& newPid);
- void updateConnection(const HttpConnection& newHttp);
+ void updateConnection(
+ const StreamingHttpConnection<v1::scheduler::Event>& newHttp);
// Closes the HTTP connection and stops the heartbeat.
//
@@ -2488,7 +2383,7 @@ struct Framework
// (scheduler driver). At most one of `http` and `pid` will be set
// according to the last connection made by the framework; neither
// field will be set if the framework is in state `RECOVERED`.
- Option<HttpConnection> http;
+ Option<StreamingHttpConnection<v1::scheduler::Event>> http;
Option<process::UPID> pid;
State state;
@@ -2569,8 +2464,8 @@ struct Framework
hashmap<SlaveID, Resources> offeredResources;
// This is only set for HTTP frameworks.
- Option<process::Owned<
- Heartbeater<mesos::scheduler::Event, v1::scheduler::Event>>> heartbeater;
+ process::Owned<ResponseHeartbeater<scheduler::Event, v1::scheduler::Event>>
+ heartbeater;
// This is used for per-framework metrics.
FrameworkMetrics metrics;
diff --git a/src/slave/http.cpp b/src/slave/http.cpp
index ba38926..4b0347f 100644
--- a/src/slave/http.cpp
+++ b/src/slave/http.cpp
@@ -831,7 +831,9 @@ Future<Response> Http::executor(
ok.type = Response::PIPE;
ok.reader = pipe.reader();
- HttpConnection http {pipe.writer(), acceptType};
+ StreamingHttpConnection<v1::executor::Event> http(
+ pipe.writer(), acceptType);
+
slave->subscribe(http, call.subscribe(), framework, executor);
return ok;
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 8467a60..2d79ee6 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -4631,7 +4631,7 @@ void Slave::operationStatusAcknowledgement(
void Slave::subscribe(
- HttpConnection http,
+ StreamingHttpConnection<v1::executor::Event> http,
const Call::Subscribe& subscribe,
Framework* framework,
Executor* executor)
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index d57aed8..469f682 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -110,7 +110,6 @@ class TaskStatusUpdateManager;
class Executor;
class Framework;
-struct HttpConnection;
struct ResourceProvider;
@@ -273,7 +272,7 @@ public:
void reconcileOperations(const ReconcileOperationsMessage& message);
void subscribe(
- HttpConnection http,
+ StreamingHttpConnection<v1::executor::Event> http,
const executor::Call::Subscribe& subscribe,
Framework* framework,
Executor* executor);
@@ -850,40 +849,6 @@ private:
};
-// Represents the streaming HTTP connection to an executor.
-struct HttpConnection
-{
- HttpConnection(const process::http::Pipe::Writer& _writer,
- ContentType _contentType)
- : writer(_writer),
- contentType(_contentType),
- encoder(lambda::bind(serialize, contentType, lambda::_1)) {}
-
- // Converts the message to an Event before sending.
- template <typename Message>
- bool send(const Message& message)
- {
- // We need to evolve the internal 'message' into a
- // 'v1::executor::Event'.
- return writer.write(encoder.encode(evolve(message)));
- }
-
- bool close()
- {
- return writer.close();
- }
-
- process::Future<Nothing> closed() const
- {
- return writer.readerClosed();
- }
-
- process::http::Pipe::Writer writer;
- ContentType contentType;
- ::recordio::Encoder<v1::executor::Event> encoder;
-};
-
-
std::ostream& operator<<(std::ostream& stream, const Executor& executor);
@@ -1020,7 +985,7 @@ public:
// * REGISTERING None None Not known yet
// * * None Some Libprocess
// * * Some None HTTP
- Option<HttpConnection> http;
+ Option<StreamingHttpConnection<v1::executor::Event>> http;
Option<process::UPID> pid;
// Tasks can be found in one of the following four data structures: