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 2016/11/29 06:15:30 UTC
[1/6] mesos git commit: Refactored recordio::ReaderProcess to be
managed by gc.
Repository: mesos
Updated Branches:
refs/heads/master 58f63747f -> d536a135d
Refactored recordio::ReaderProcess to be managed by gc.
This ensures that callers do not accidentally wait on the `read`
future and end up waiting on ReaderProcess from within the context
of the ReaderProcess.
Review: https://reviews.apache.org/r/54093
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/e0d8057c
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/e0d8057c
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/e0d8057c
Branch: refs/heads/master
Commit: e0d8057ce123a1c325070382043608932aaabb06
Parents: 0647480
Author: Vinod Kone <vi...@gmail.com>
Authored: Wed Nov 23 15:34:45 2016 -0800
Committer: Vinod Kone <vi...@gmail.com>
Committed: Mon Nov 28 22:14:00 2016 -0800
----------------------------------------------------------------------
src/common/recordio.hpp | 26 +++++++++++++++++---------
1 file changed, 17 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/e0d8057c/src/common/recordio.hpp
----------------------------------------------------------------------
diff --git a/src/common/recordio.hpp b/src/common/recordio.hpp
index cbd9d36..4ecefd8 100644
--- a/src/common/recordio.hpp
+++ b/src/common/recordio.hpp
@@ -27,6 +27,7 @@
#include <process/dispatch.hpp>
#include <process/http.hpp>
#include <process/owned.hpp>
+#include <process/pid.hpp>
#include <process/process.hpp>
#include <stout/lambda.hpp>
@@ -59,17 +60,23 @@ template <typename T>
class Reader
{
public:
+ // We spawn `ReaderProcess` as a managed process to guarantee
+ // that it does not wait on itself (this would cause a deadlock!).
+ // See comments in `Connection::Data` for further details.
Reader(::recordio::Decoder<T>&& decoder,
process::http::Pipe::Reader reader)
- : process(new internal::ReaderProcess<T>(std::move(decoder), reader))
- {
- process::spawn(process.get());
- }
+ : process(process::spawn(
+ new internal::ReaderProcess<T>(std::move(decoder), reader),
+ true)) {}
virtual ~Reader()
{
- process::terminate(process.get());
- process::wait(process.get());
+ // Note that we pass 'false' here to avoid injecting the
+ // termination event at the front of the queue. This is
+ // to ensure we don't drop any queued request dispatches
+ // which would leave the caller with a future stuck in
+ // a pending state.
+ process::terminate(process, false);
}
/**
@@ -80,11 +87,11 @@ public:
*/
process::Future<Result<T>> read()
{
- return process::dispatch(process.get(), &internal::ReaderProcess<T>::read);
+ return process::dispatch(process, &internal::ReaderProcess<T>::read);
}
private:
- process::Owned<internal::ReaderProcess<T>> process;
+ process::PID<internal::ReaderProcess<T>> process;
};
@@ -97,7 +104,8 @@ public:
ReaderProcess(
::recordio::Decoder<T>&& _decoder,
process::http::Pipe::Reader _reader)
- : decoder(_decoder),
+ : process::ProcessBase(process::ID::generate("__reader__")),
+ decoder(_decoder),
reader(_reader),
done(false) {}
[3/6] mesos git commit: Added streaming support to `/api/v1` handler
on the agent.
Posted by vi...@apache.org.
Added streaming support to `/api/v1` handler on the agent.
Note that this change only updates the handler to correctly
handle non-streaming calls given it receives `PIPE` requests.
Handling streaming calls will come later.
Review: https://reviews.apache.org/r/53994
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/ad786c58
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/ad786c58
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/ad786c58
Branch: refs/heads/master
Commit: ad786c5831bd0adb199c701f6aeef482623c5ffb
Parents: 95030c5
Author: Vinod Kone <vi...@gmail.com>
Authored: Tue Nov 15 20:09:17 2016 -0800
Committer: Vinod Kone <vi...@gmail.com>
Committed: Mon Nov 28 22:14:01 2016 -0800
----------------------------------------------------------------------
src/slave/http.cpp | 85 +++++++++++++++++++++++++++++++-----------------
src/slave/slave.cpp | 7 ++--
src/slave/slave.hpp | 7 ++++
3 files changed, 68 insertions(+), 31 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/ad786c58/src/slave/http.cpp
----------------------------------------------------------------------
diff --git a/src/slave/http.cpp b/src/slave/http.cpp
index aa9f492..cca88b8 100644
--- a/src/slave/http.cpp
+++ b/src/slave/http.cpp
@@ -310,47 +310,42 @@ Future<Response> Slave::Http::api(
return MethodNotAllowed({"POST"}, request.method);
}
- v1::agent::Call v1Call;
-
- Option<string> contentType = request.headers.get("Content-Type");
- if (contentType.isNone()) {
+ 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::agent::Call> parse =
- ::protobuf::parse<v1::agent::Call>(value.get());
-
- if (parse.isError()) {
- return BadRequest("Failed to convert JSON into Call protobuf: " +
- parse.error());
- }
-
- v1Call = parse.get();
+ ContentType contentType;
+ if (contentType_.get() == APPLICATION_JSON) {
+ contentType = ContentType::JSON;
+ } else if (contentType_.get() == APPLICATION_PROTOBUF) {
+ contentType = ContentType::PROTOBUF;
} else {
return UnsupportedMediaType(
string("Expecting 'Content-Type' of ") +
APPLICATION_JSON + " or " + APPLICATION_PROTOBUF);
}
- agent::Call call = devolve(v1Call);
+ // This lambda deserializes a string into a valid `Call`
+ // based on the content type.
+ auto deserializer = [](const string& body, ContentType contentType)
+ -> Try<agent::Call> {
+ Try<v1::agent::Call> v1Call =
+ deserialize<v1::agent::Call>(contentType, body);
+
+ if (v1Call.isError()) {
+ return Error(v1Call.error());
+ }
- Option<Error> error = validation::agent::call::validate(call);
+ agent::Call call = devolve(v1Call.get());
- if (error.isSome()) {
- return BadRequest("Failed to validate agent::Call: " + error.get().message);
- }
+ Option<Error> error = validation::agent::call::validate(call);
+ if (error.isSome()) {
+ return Error("Failed to validate agent::Call: " + error.get().message);
+ }
- LOG(INFO) << "Processing call " << call.type();
+ return call;
+ };
ContentType acceptType;
if (request.acceptsMediaType(APPLICATION_JSON)) {
@@ -363,6 +358,38 @@ Future<Response> Slave::Http::api(
"'" + APPLICATION_PROTOBUF + "' or '" + APPLICATION_JSON + "'");
}
+ if (contentType == ContentType::JSON ||
+ contentType == ContentType::PROTOBUF) {
+ CHECK_EQ(http::Request::PIPE, request.type);
+ CHECK_SOME(request.reader);
+
+ Pipe::Reader reader = request.reader.get(); // Remove const.
+
+ return reader.readAll()
+ .then(defer(
+ slave->self(),
+ [=](const string& body) -> Future<Response> {
+ Try<agent::Call> call = deserializer(body, contentType);
+ if (call.isError()) {
+ return BadRequest(call.error());
+ }
+ return _api(call.get(), contentType, acceptType, principal);
+ }));
+ } else {
+ // TODO(vinod): Add support for 'streaming' content type.
+ UNREACHABLE();
+ }
+}
+
+
+Future<Response> Slave::Http::_api(
+ const agent::Call& call,
+ ContentType contentType,
+ ContentType acceptType,
+ const Option<string>& principal) const
+{
+ LOG(INFO) << "Processing call " << call.type();
+
switch (call.type()) {
case agent::Call::UNKNOWN:
return NotImplemented();
http://git-wip-us.apache.org/repos/asf/mesos/blob/ad786c58/src/slave/slave.cpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.cpp b/src/slave/slave.cpp
index 521f08d..1273b0f 100644
--- a/src/slave/slave.cpp
+++ b/src/slave/slave.cpp
@@ -648,7 +648,9 @@ void Slave::initialize()
&Slave::ping,
&PingSlaveMessage::connected);
-
+ // Setup the '/api/v1' handler for streaming requests.
+ RouteOptions options;
+ options.requestStreaming = true;
route("/api/v1",
// TODO(benh): Is this authentication realm sufficient or do
// we need some kind of hybrid if we expect both executors
@@ -659,7 +661,8 @@ void Slave::initialize()
const Option<string>& principal) {
Http::log(request);
return http.api(request, principal);
- });
+ },
+ options);
route("/api/v1/executor",
Http::EXECUTOR_HELP(),
http://git-wip-us.apache.org/repos/asf/mesos/blob/ad786c58/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index 0cc1054..0ede185 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -523,6 +523,13 @@ private:
private:
JSON::Object _flags() const;
+ // Continuation for `/api` endpoint that handles non-streaming requests.
+ process::Future<process::http::Response> _api(
+ const agent::Call& call,
+ ContentType contentTye,
+ ContentType acceptType,
+ const Option<std::string>& principal) const;
+
// Make continuation for `statistics` `static` as it might
// execute when the invoking `Http` is already destructed.
process::http::Response _statistics(
[4/6] mesos git commit: Aliased `process::http` namespace to `http`.
Posted by vi...@apache.org.
Aliased `process::http` namespace to `http`.
This clarifies usage of objects from different namespaces,
e.g., http::Request vs mesos::Request.
Review: https://reviews.apache.org/r/54049
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/e1be7cf6
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/e1be7cf6
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/e1be7cf6
Branch: refs/heads/master
Commit: e1be7cf6bdc2a7c1a8f36d0dac270e4152400da1
Parents: ad786c5
Author: Vinod Kone <vi...@gmail.com>
Authored: Wed Nov 23 14:09:59 2016 -0800
Committer: Vinod Kone <vi...@gmail.com>
Committed: Mon Nov 28 22:14:01 2016 -0800
----------------------------------------------------------------------
src/tests/api_tests.cpp | 188 +++++++++++++++++++++----------------------
1 file changed, 91 insertions(+), 97 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/e1be7cf6/src/tests/api_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/api_tests.cpp b/src/tests/api_tests.cpp
index 8889e78..03e8590 100644
--- a/src/tests/api_tests.cpp
+++ b/src/tests/api_tests.cpp
@@ -53,6 +53,8 @@
#include "tests/containerizer/mock_containerizer.hpp"
+namespace http = process::http;
+
using mesos::master::detector::MasterDetector;
using mesos::master::detector::StandaloneMasterDetector;
@@ -75,14 +77,6 @@ using process::Future;
using process::Owned;
using process::Promise;
-using process::http::Accepted;
-using process::http::BadRequest;
-using process::http::NotFound;
-using process::http::NotImplemented;
-using process::http::OK;
-using process::http::Pipe;
-using process::http::Response;
-
using recordio::Decoder;
using testing::_;
@@ -108,18 +102,18 @@ public:
const v1::master::Call& call,
const ContentType& contentType)
{
- process::http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
+ http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
headers["Accept"] = stringify(contentType);
- return process::http::post(
+ return http::post(
pid,
"api/v1",
headers,
serialize(contentType, call),
stringify(contentType))
- .then([contentType](const Response& response)
+ .then([contentType](const http::Response& response)
-> Future<v1::master::Response> {
- if (response.status != OK().status) {
+ if (response.status != http::OK().status) {
return Failure("Unexpected response status " + response.status);
}
return deserialize<v1::master::Response>(contentType, response.body);
@@ -748,17 +742,17 @@ TEST_P(MasterAPITest, SetLoggingLevel)
setLoggingLevel->mutable_duration()->set_nanoseconds(toggleDuration.ns());
ContentType contentType = GetParam();
- process::http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
+ http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
headers["Accept"] = stringify(contentType);
- Future<Nothing> v1Response = process::http::post(
+ Future<Nothing> v1Response = http::post(
master.get()->pid,
"api/v1",
headers,
serialize(contentType, v1Call),
stringify(contentType))
- .then([contentType](const Response& response) -> Future<Nothing> {
- if (response.status != OK().status) {
+ .then([contentType](const http::Response& response) -> Future<Nothing> {
+ if (response.status != http::OK().status) {
return Failure("Unexpected response status " + response.status);
}
return Nothing();
@@ -828,14 +822,14 @@ TEST_P(MasterAPITest, ListFilesInvalidPath)
ContentType contentType = GetParam();
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
master.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(NotFound().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::NotFound().status, response);
}
@@ -987,14 +981,14 @@ TEST_P(MasterAPITest, ReserveResources)
ContentType contentType = GetParam();
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
master.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(Accepted().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::Accepted().status, response);
AWAIT_READY(offers);
@@ -1051,14 +1045,14 @@ TEST_P(MasterAPITest, UnreserveResources)
ContentType contentType = GetParam();
- Future<Response> reserveResponse = process::http::post(
+ Future<http::Response> reserveResponse = http::post(
master.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(Accepted().status, reserveResponse);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::Accepted().status, reserveResponse);
MockScheduler sched;
MesosSchedulerDriver driver(
@@ -1096,14 +1090,14 @@ TEST_P(MasterAPITest, UnreserveResources)
unreserveResources->mutable_resources()->CopyFrom(
evolve(dynamicallyReserved));
- Future<Response> unreserveResponse = process::http::post(
+ Future<http::Response> unreserveResponse = http::post(
master.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(Accepted().status, unreserveResponse);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::Accepted().status, unreserveResponse);
AWAIT_READY(offers);
@@ -1126,7 +1120,7 @@ TEST_P(MasterAPITest, UpdateAndGetMaintenanceSchedule)
ASSERT_SOME(master);
ContentType contentType = GetParam();
- process::http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
+ http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
headers["Accept"] = stringify(contentType);
// Generate `MachineID`s that can be used in this test.
@@ -1146,14 +1140,14 @@ TEST_P(MasterAPITest, UpdateAndGetMaintenanceSchedule)
v1UpdateScheduleCall.mutable_update_maintenance_schedule();
maintenanceSchedule->mutable_schedule()->CopyFrom(v1Schedule);
- Future<Nothing> v1UpdateScheduleResponse = process::http::post(
+ Future<Nothing> v1UpdateScheduleResponse = http::post(
master.get()->pid,
"api/v1",
headers,
serialize(contentType, v1UpdateScheduleCall),
stringify(contentType))
- .then([contentType](const Response& response) -> Future<Nothing> {
- if (response.status != OK().status) {
+ .then([contentType](const http::Response& response) -> Future<Nothing> {
+ if (response.status != http::OK().status) {
return Failure("Unexpected response status " + response.status);
}
return Nothing();
@@ -1192,7 +1186,7 @@ TEST_P(MasterAPITest, GetMaintenanceStatus)
ASSERT_SOME(master);
ContentType contentType = GetParam();
- process::http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
+ http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
headers["Accept"] = stringify(contentType);
// Generate `MachineID`s that can be used in this test.
@@ -1212,14 +1206,14 @@ TEST_P(MasterAPITest, GetMaintenanceStatus)
v1UpdateScheduleCall.mutable_update_maintenance_schedule();
maintenanceSchedule->mutable_schedule()->CopyFrom(v1Schedule);
- Future<Nothing> v1UpdateScheduleResponse = process::http::post(
+ Future<Nothing> v1UpdateScheduleResponse = http::post(
master.get()->pid,
"api/v1",
headers,
serialize(contentType, v1UpdateScheduleCall),
stringify(contentType))
- .then([contentType](const Response& response) -> Future<Nothing> {
- if (response.status != OK().status) {
+ .then([contentType](const http::Response& response) -> Future<Nothing> {
+ if (response.status != http::OK().status) {
return Failure("Unexpected response status " + response.status);
}
return Nothing();
@@ -1258,7 +1252,7 @@ TEST_P(MasterAPITest, StartAndStopMaintenance)
ASSERT_SOME(master);
ContentType contentType = GetParam();
- process::http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
+ http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
headers["Accept"] = stringify(contentType);
// Generate `MachineID`s that can be used in this test.
@@ -1284,14 +1278,14 @@ TEST_P(MasterAPITest, StartAndStopMaintenance)
v1UpdateScheduleCall.mutable_update_maintenance_schedule();
maintenanceSchedule->mutable_schedule()->CopyFrom(v1Schedule);
- Future<Nothing> v1UpdateScheduleResponse = process::http::post(
+ Future<Nothing> v1UpdateScheduleResponse = http::post(
master.get()->pid,
"api/v1",
headers,
serialize(contentType, v1UpdateScheduleCall),
stringify(contentType))
- .then([contentType](const Response& response) -> Future<Nothing> {
- if (response.status != OK().status) {
+ .then([contentType](const http::Response& response) -> Future<Nothing> {
+ if (response.status != http::OK().status) {
return Failure("Unexpected response status " + response.status);
}
return Nothing();
@@ -1306,14 +1300,14 @@ TEST_P(MasterAPITest, StartAndStopMaintenance)
v1StartMaintenanceCall.mutable_start_maintenance();
startMaintenance->add_machines()->CopyFrom(evolve(machine3));
- Future<Nothing> v1StartMaintenanceResponse = process::http::post(
+ Future<Nothing> v1StartMaintenanceResponse = http::post(
master.get()->pid,
"api/v1",
headers,
serialize(contentType, v1StartMaintenanceCall),
stringify(contentType))
- .then([contentType](const Response& response) -> Future<Nothing> {
- if (response.status != OK().status) {
+ .then([contentType](const http::Response& response) -> Future<Nothing> {
+ if (response.status != http::OK().status) {
return Failure("Unexpected response status " + response.status);
}
return Nothing();
@@ -1328,14 +1322,14 @@ TEST_P(MasterAPITest, StartAndStopMaintenance)
v1StopMaintenanceCall.mutable_stop_maintenance();
stopMaintenance->add_machines()->CopyFrom(evolve(machine3));
- Future<Nothing> v1StopMaintenanceResponse = process::http::post(
+ Future<Nothing> v1StopMaintenanceResponse = http::post(
master.get()->pid,
"api/v1",
headers,
serialize(contentType, v1StopMaintenanceCall),
stringify(contentType))
- .then([contentType](const Response& response) -> Future<Nothing> {
- if (response.status != OK().status) {
+ .then([contentType](const http::Response& response) -> Future<Nothing> {
+ if (response.status != http::OK().status) {
return Failure("Unexpected response status " + response.status);
}
return Nothing();
@@ -1378,23 +1372,23 @@ TEST_P(MasterAPITest, SubscribeAgentEvents)
v1::master::Call v1Call;
v1Call.set_type(v1::master::Call::SUBSCRIBE);
- process::http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
+ http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
headers["Accept"] = stringify(contentType);
- Future<Response> response = process::http::streaming::post(
+ Future<http::Response> response = http::streaming::post(
master.get()->pid,
"api/v1",
headers,
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response);
AWAIT_EXPECT_RESPONSE_HEADER_EQ("chunked", "Transfer-Encoding", response);
- ASSERT_EQ(Response::PIPE, response.get().type);
+ ASSERT_EQ(http::Response::PIPE, response.get().type);
ASSERT_SOME(response->reader);
- Pipe::Reader reader = response->reader.get();
+ http::Pipe::Reader reader = response->reader.get();
auto deserializer =
lambda::bind(deserialize<v1::master::Event>, contentType, lambda::_1);
@@ -1525,23 +1519,23 @@ TEST_P(MasterAPITest, Subscribe)
v1::master::Call v1Call;
v1Call.set_type(v1::master::Call::SUBSCRIBE);
- process::http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
+ http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
headers["Accept"] = stringify(contentType);
- Future<Response> response = process::http::streaming::post(
+ Future<http::Response> response = http::streaming::post(
master.get()->pid,
"api/v1",
headers,
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response);
AWAIT_EXPECT_RESPONSE_HEADER_EQ("chunked", "Transfer-Encoding", response);
- ASSERT_EQ(Response::PIPE, response.get().type);
+ ASSERT_EQ(http::Response::PIPE, response.get().type);
ASSERT_SOME(response->reader);
- Pipe::Reader reader = response->reader.get();
+ http::Pipe::Reader reader = response->reader.get();
auto deserializer =
lambda::bind(deserialize<v1::master::Event>, contentType, lambda::_1);
@@ -1672,14 +1666,14 @@ TEST_P(MasterAPITest, GetQuota)
quotaRequest->mutable_guarantee()->CopyFrom(quotaResources);
// Send a quota request for the specified role.
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
master.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response);
}
// Verify the quota is set using the `GET_QUOTA` call.
@@ -1723,14 +1717,14 @@ TEST_P(MasterAPITest, SetQuota)
ContentType contentType = GetParam();
// Send a quota request for the specified role.
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
master.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response);
}
@@ -1760,14 +1754,14 @@ TEST_P(MasterAPITest, RemoveQuota)
ContentType contentType = GetParam();
// Send a quota request for the specified role.
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
master.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response);
}
// Verify if the quota is set using `GET_QUOTA` call.
@@ -1797,14 +1791,14 @@ TEST_P(MasterAPITest, RemoveQuota)
ContentType contentType = GetParam();
// Send a quota request for the specified role.
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
master.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response);
}
// Verify if the quota is removed using `GET_QUOTA` call.
@@ -1876,17 +1870,17 @@ TEST_P(MasterAPITest, CreateAndDestroyVolumes)
ContentType contentType = GetParam();
- process::http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
+ http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
headers["Accept"] = stringify(contentType);
- Future<Nothing> v1CreateVolumesResponse = process::http::post(
+ Future<Nothing> v1CreateVolumesResponse = http::post(
master.get()->pid,
"api/v1",
headers,
serialize(contentType, v1CreateVolumesCall),
stringify(contentType))
- .then([contentType](const Response& response) -> Future<Nothing> {
- if (response.status != Accepted().status) {
+ .then([contentType](const http::Response& response) -> Future<Nothing> {
+ if (response.status != http::Accepted().status) {
return Failure("Unexpected response status " + response.status);
}
return Nothing();
@@ -1952,14 +1946,14 @@ TEST_P(MasterAPITest, CreateAndDestroyVolumes)
destroyVolumes->mutable_agent_id()->CopyFrom(evolve(slaveId));
destroyVolumes->add_volumes()->CopyFrom(evolve(volume));
- Future<Nothing> v1DestroyVolumesResponse = process::http::post(
+ Future<Nothing> v1DestroyVolumesResponse = http::post(
master.get()->pid,
"api/v1",
headers,
serialize(contentType, v1DestroyVolumesCall),
stringify(contentType))
- .then([contentType](const Response& response) -> Future<Nothing> {
- if (response.status != Accepted().status) {
+ .then([contentType](const http::Response& response) -> Future<Nothing> {
+ if (response.status != http::Accepted().status) {
return Failure("Unexpected response status " + response.status);
}
return Nothing();
@@ -2029,17 +2023,17 @@ TEST_P(MasterAPITest, UpdateWeights)
weightInfo->set_role("role");
weightInfo->set_weight(4.0);
- process::http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
+ http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
headers["Accept"] = stringify(contentType);
- Future<Nothing> updateResponse = process::http::post(
+ Future<Nothing> updateResponse = http::post(
master.get()->pid,
"api/v1",
headers,
serialize(contentType, updateCall),
stringify(contentType))
- .then([contentType](const Response& response) -> Future<Nothing> {
- if (response.status != OK().status) {
+ .then([contentType](const http::Response& response) -> Future<Nothing> {
+ if (response.status != http::OK().status) {
return Failure("Unexpected response status " + response.status);
}
return Nothing();
@@ -2181,14 +2175,14 @@ TEST_P(MasterAPITest, ReadFileInvalidPath)
ContentType contentType = GetParam();
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
master.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(NotFound().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::NotFound().status, response);
}
@@ -2204,18 +2198,18 @@ public:
const v1::agent::Call& call,
const ContentType& contentType)
{
- process::http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
+ http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
headers["Accept"] = stringify(contentType);
- return process::http::post(
+ return http::post(
pid,
"api/v1",
headers,
serialize(contentType, call),
stringify(contentType))
- .then([contentType](const Response& response)
+ .then([contentType](const http::Response& response)
-> Future<v1::agent::Response> {
- if (response.status != OK().status) {
+ if (response.status != http::OK().status) {
return Failure("Unexpected response status " + response.status);
}
return deserialize<v1::agent::Response>(contentType, response.body);
@@ -2425,17 +2419,17 @@ TEST_P(AgentAPITest, SetLoggingLevel)
setLoggingLevel->mutable_duration()->set_nanoseconds(toggleDuration.ns());
ContentType contentType = GetParam();
- process::http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
+ http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL);
headers["Accept"] = stringify(contentType);
- Future<Nothing> v1Response = process::http::post(
+ Future<Nothing> v1Response = http::post(
slave.get()->pid,
"api/v1",
headers,
serialize(contentType, v1Call),
stringify(contentType))
- .then([contentType](const Response& response) -> Future<Nothing> {
- if (response.status != OK().status) {
+ .then([contentType](const http::Response& response) -> Future<Nothing> {
+ if (response.status != http::OK().status) {
return Failure("Unexpected response status " + response.status);
}
return Nothing();
@@ -2523,14 +2517,14 @@ TEST_P(AgentAPITest, ListFilesInvalidPath)
ContentType contentType = GetParam();
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
slave.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(NotFound().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::NotFound().status, response);
}
@@ -2780,14 +2774,14 @@ TEST_P(AgentAPITest, ReadFileInvalidPath)
ContentType contentType = GetParam();
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
slave.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, v1Call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(NotFound().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::NotFound().status, response);
}
@@ -3309,14 +3303,14 @@ TEST_P(AgentAPITest, NestedContainerWaitNotFound)
call.mutable_wait_nested_container()->mutable_container_id()
->CopyFrom(unknownContainerId);
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
slave.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(NotFound().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::NotFound().status, response);
}
// The destructor of `cluster::Slave` will try to clean up any
@@ -3364,14 +3358,14 @@ TEST_P(AgentAPITest, NestedContainerKillNotFound)
call.mutable_kill_nested_container()->mutable_container_id()
->CopyFrom(unknownContainerId);
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
slave.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(NotFound().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::NotFound().status, response);
}
// The destructor of `cluster::Slave` will try to clean up any
@@ -3446,14 +3440,14 @@ TEST_P(AgentAPITest, NestedContainerLaunchFalse)
call.mutable_launch_nested_container()->mutable_container_id()
->CopyFrom(containerId);
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
slave.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(BadRequest().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::BadRequest().status, response);
}
EXPECT_CALL(exec, shutdown(_))
@@ -3495,14 +3489,14 @@ TEST_P(AgentAPITest, NestedContainerLaunchGrandchildNotSupported)
call.mutable_launch_nested_container()->mutable_container_id()
->CopyFrom(containerId);
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
slave.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(NotImplemented().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::NotImplemented().status, response);
}
}
@@ -3565,14 +3559,14 @@ TEST_P(AgentAPITest, NestedContainerLaunch)
call.mutable_launch_nested_container()->mutable_container_id()
->CopyFrom(containerId);
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
slave.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response);
}
Future<v1::agent::Response> wait;
@@ -3599,14 +3593,14 @@ TEST_P(AgentAPITest, NestedContainerLaunch)
call.mutable_kill_nested_container()->mutable_container_id()
->CopyFrom(containerId);
- Future<Response> response = process::http::post(
+ Future<http::Response> response = http::post(
slave.get()->pid,
"api/v1",
createBasicAuthHeaders(DEFAULT_CREDENTIAL),
serialize(contentType, call),
stringify(contentType));
- AWAIT_EXPECT_RESPONSE_STATUS_EQ(OK().status, response);
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response);
}
AWAIT_READY(wait);
[2/6] mesos git commit: Fixed a bug in scheduler library.
Posted by vi...@apache.org.
Fixed a bug in scheduler library.
This bug fixes handling of EOF and deserialization errors.
Review: https://reviews.apache.org/r/54048
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/06474801
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/06474801
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/06474801
Branch: refs/heads/master
Commit: 064748011705e066eab7d764fb3522b6f3979974
Parents: 58f6374
Author: Vinod Kone <vi...@gmail.com>
Authored: Sat Nov 26 09:35:21 2016 -0800
Committer: Vinod Kone <vi...@gmail.com>
Committed: Mon Nov 28 22:14:00 2016 -0800
----------------------------------------------------------------------
src/scheduler/scheduler.cpp | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/06474801/src/scheduler/scheduler.cpp
----------------------------------------------------------------------
diff --git a/src/scheduler/scheduler.cpp b/src/scheduler/scheduler.cpp
index 4b6fae1..d545c9f 100644
--- a/src/scheduler/scheduler.cpp
+++ b/src/scheduler/scheduler.cpp
@@ -642,7 +642,7 @@ protected:
}
// This could happen if the master failed over after sending an event.
- if (!event->isSome()) {
+ if (event->isNone()) {
const string error = "End-Of-File received from master. The master "
"closed the event stream";
LOG(ERROR) << error;
[6/6] mesos git commit: Added `recordio::transform` helper.
Posted by vi...@apache.org.
Added `recordio::transform` helper.
This helper lets us transform raw bytes read from a RecordIO
reader and write to a pipe.
Review: https://reviews.apache.org/r/54039
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/95030c59
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/95030c59
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/95030c59
Branch: refs/heads/master
Commit: 95030c59d169621027c3c1e055ae1a6d0b3d78b5
Parents: e0d8057
Author: Vinod Kone <vi...@gmail.com>
Authored: Tue Nov 22 16:58:16 2016 -0800
Committer: Vinod Kone <vi...@gmail.com>
Committed: Mon Nov 28 22:14:01 2016 -0800
----------------------------------------------------------------------
src/common/recordio.hpp | 41 +++++++++++
src/tests/common/recordio_tests.cpp | 113 +++++++++++++++++++++++++++++++
2 files changed, 154 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/95030c59/src/common/recordio.hpp
----------------------------------------------------------------------
diff --git a/src/common/recordio.hpp b/src/common/recordio.hpp
index 4ecefd8..0f6b47b 100644
--- a/src/common/recordio.hpp
+++ b/src/common/recordio.hpp
@@ -31,6 +31,7 @@
#include <process/process.hpp>
#include <stout/lambda.hpp>
+#include <stout/nothing.hpp>
#include <stout/recordio.hpp>
#include <stout/result.hpp>
@@ -95,6 +96,46 @@ private:
};
+/**
+ * This is a helper function that reads records from a `Reader`, applies
+ * a transformation to the records and writes to the pipe.
+ *
+ * Returns a failed future if there are any errors reading or writing.
+ * The future is satisfied when we get a EOF.
+ *
+ * TODO(vinod): Split this method into primitives that can transform a
+ * stream of bytes to a stream of typed records that can be further transformed.
+ * See the TODO above in `Reader` for further details.
+ */
+template <typename T>
+process::Future<Nothing> transform(
+ process::Owned<Reader<T>>&& reader,
+ const std::function<std::string(const T&)>& func,
+ process::http::Pipe::Writer writer)
+{
+ return reader->read()
+ .then([=](const Result<T>& record) mutable -> process::Future<Nothing> {
+ // This could happen if EOF is sent by the writer.
+ if (record.isNone()) {
+ return Nothing();
+ }
+
+ // This could happen if there is a de-serialization error.
+ if (record.isError()) {
+ return process::Failure(record.error());
+ }
+
+ // TODO(vinod): Instead of detecting that the reader went away only
+ // after attempting a write, leverage `writer.readerClosed` future.
+ if (!writer.write(func(record.get()))) {
+ return process::Failure("Write failed to the pipe");
+ }
+
+ return transform(std::move(reader), func, writer);
+ });
+}
+
+
namespace internal {
template <typename T>
http://git-wip-us.apache.org/repos/asf/mesos/blob/95030c59/src/tests/common/recordio_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/common/recordio_tests.cpp b/src/tests/common/recordio_tests.cpp
index 86a1117..872a1dc 100644
--- a/src/tests/common/recordio_tests.cpp
+++ b/src/tests/common/recordio_tests.cpp
@@ -159,3 +159,116 @@ TEST(RecordIOReaderTest, PipeFailure)
// Subsequent reads should return a failure.
AWAIT_EXPECT_FAILED(reader.read());
}
+
+
+// This test verifies that when an EOF is received by the `writer` used
+// in `transform`, the future returned to the caller is satisfied.
+TEST(RecordIOTransformTest, EndOfFile)
+{
+ // Write some data to the pipe so that records
+ // are available before any reads occur.
+ ::recordio::Encoder<string> encoder(strings::upper);
+
+ string data;
+
+ data += encoder.encode("hello ");
+ data += encoder.encode("world! ");
+
+ process::http::Pipe pipeA;
+ pipeA.writer().write(data);
+
+ process::Owned<internal::recordio::Reader<string>> reader(
+ new internal::recordio::Reader<string>(
+ ::recordio::Decoder<string>(strings::lower),
+ pipeA.reader()));
+
+ process::http::Pipe pipeB;
+
+ auto trim = [](const string& str) { return strings::trim(str); };
+
+ Future<Nothing> transform = internal::recordio::transform<string>(
+ std::move(reader), trim, pipeB.writer());
+
+ Future<string> future = pipeB.reader().readAll();
+
+ pipeA.writer().close();
+
+ AWAIT_READY(transform);
+
+ pipeB.writer().close();
+
+ AWAIT_ASSERT_EQ("helloworld!", future);
+}
+
+
+// This test verifies that when the write end of the `reader` used in
+// `transform` fails, a failure is returned to the caller.
+TEST(RecordIOTransformTest, ReaderWriterEndFail)
+{
+ // Write some data to the pipe so that records
+ // are available before any reads occur.
+ ::recordio::Encoder<string> encoder(strings::upper);
+
+ string data;
+
+ data += encoder.encode("hello ");
+ data += encoder.encode("world! ");
+
+ process::http::Pipe pipeA;
+ pipeA.writer().write(data);
+
+ process::Owned<internal::recordio::Reader<string>> reader(
+ new internal::recordio::Reader<string>(
+ ::recordio::Decoder<string>(strings::lower),
+ pipeA.reader()));
+
+ process::http::Pipe pipeB;
+
+ auto trim = [](const string& str) { return strings::trim(str); };
+
+ Future<Nothing> transform = internal::recordio::transform<string>(
+ std::move(reader), trim, pipeB.writer());
+
+ Future<string> future = pipeB.reader().readAll();
+
+ pipeA.writer().fail("Writer failure");
+
+ AWAIT_FAILED(transform);
+ ASSERT_TRUE(future.isPending());
+}
+
+
+// This test verifies that when the read end of the `writer` used in
+// `transform` is closed, a failure is returned to the caller.
+TEST(RecordIOTransformTest, WriterReadEndFail)
+{
+ // Write some data to the pipe so that records
+ // are available before any reads occur.
+ ::recordio::Encoder<string> encoder(strings::upper);
+
+ string data;
+
+ data += encoder.encode("hello ");
+ data += encoder.encode("world! ");
+
+ process::http::Pipe pipeA;
+ pipeA.writer().write(data);
+
+ process::Owned<internal::recordio::Reader<string>> reader(
+ new internal::recordio::Reader<string>(
+ ::recordio::Decoder<string>(strings::lower),
+ pipeA.reader()));
+
+ process::http::Pipe pipeB;
+
+ auto trim = [](const string& str) { return strings::trim(str); };
+
+ pipeB.reader().close();
+
+ Future<Nothing> transform = internal::recordio::transform<string>(
+ std::move(reader), trim, pipeB.writer());
+
+ pipeA.writer().close();
+
+ AWAIT_FAILED(transform);
+}
[5/6] mesos git commit: Added API handler for ATTACH_CONTAINER_OUTPUT.
Posted by vi...@apache.org.
Added API handler for ATTACH_CONTAINER_OUTPUT.
Review: https://reviews.apache.org/r/53995
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/d536a135
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/d536a135
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/d536a135
Branch: refs/heads/master
Commit: d536a135dfee0408a5b7de8e02c449f8b69edbfc
Parents: e1be7cf
Author: Vinod Kone <vi...@gmail.com>
Authored: Sun Nov 20 22:27:26 2016 +0800
Committer: Vinod Kone <vi...@gmail.com>
Committed: Mon Nov 28 22:14:01 2016 -0800
----------------------------------------------------------------------
src/internal/evolve.cpp | 6 +++
src/internal/evolve.hpp | 1 +
src/slave/http.cpp | 101 +++++++++++++++++++++++++++++++++++++++++-
src/slave/slave.hpp | 6 +++
src/slave/validation.cpp | 17 ++++++-
src/tests/api_tests.cpp | 37 ++++++++++++++++
6 files changed, 166 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/d536a135/src/internal/evolve.cpp
----------------------------------------------------------------------
diff --git a/src/internal/evolve.cpp b/src/internal/evolve.cpp
index 23a515c..5ef2306 100644
--- a/src/internal/evolve.cpp
+++ b/src/internal/evolve.cpp
@@ -195,6 +195,12 @@ v1::agent::Call evolve(const mesos::agent::Call& call)
}
+v1::agent::ProcessIO evolve(const mesos::agent::ProcessIO& processIO)
+{
+ return evolve<v1::agent::ProcessIO>(processIO);
+}
+
+
v1::agent::Response evolve(const mesos::agent::Response& response)
{
return evolve<v1::agent::Response>(response);
http://git-wip-us.apache.org/repos/asf/mesos/blob/d536a135/src/internal/evolve.hpp
----------------------------------------------------------------------
diff --git a/src/internal/evolve.hpp b/src/internal/evolve.hpp
index e8b03c9..73d55eb 100644
--- a/src/internal/evolve.hpp
+++ b/src/internal/evolve.hpp
@@ -76,6 +76,7 @@ v1::TaskInfo evolve(const TaskInfo& taskInfo);
v1::TaskStatus evolve(const TaskStatus& status);
v1::agent::Call evolve(const mesos::agent::Call& call);
+v1::agent::ProcessIO evolve(const mesos::agent::ProcessIO& processIO);
v1::agent::Response evolve(const mesos::agent::Response& response);
v1::maintenance::ClusterStatus evolve(
http://git-wip-us.apache.org/repos/asf/mesos/blob/d536a135/src/slave/http.cpp
----------------------------------------------------------------------
diff --git a/src/slave/http.cpp b/src/slave/http.cpp
index cca88b8..91bb882 100644
--- a/src/slave/http.cpp
+++ b/src/slave/http.cpp
@@ -55,6 +55,7 @@
#include "common/build.hpp"
#include "common/http.hpp"
+#include "common/recordio.hpp"
#include "internal/devolve.hpp"
@@ -66,6 +67,10 @@
#include "version/version.hpp"
+using mesos::agent::ProcessIO;
+
+using mesos::internal::recordio::Reader;
+
using mesos::slave::ContainerTermination;
using process::AUTHENTICATION;
@@ -80,6 +85,7 @@ using process::TLDR;
using process::http::Accepted;
using process::http::BadRequest;
+using process::http::Connection;
using process::http::Forbidden;
using process::http::NotFound;
using process::http::InternalServerError;
@@ -93,6 +99,8 @@ using process::http::UnsupportedMediaType;
using process::metrics::internal::MetricsProcess;
+using ::recordio::Decoder;
+
using std::list;
using std::string;
using std::tie;
@@ -444,8 +452,10 @@ Future<Response> Slave::Http::_api(
case mesos::agent::Call::LAUNCH_NESTED_CONTAINER_SESSION:
case mesos::agent::Call::ATTACH_CONTAINER_INPUT:
- case mesos::agent::Call::ATTACH_CONTAINER_OUTPUT:
return NotImplemented();
+
+ case mesos::agent::Call::ATTACH_CONTAINER_OUTPUT:
+ return attachContainerOutput(call, contentType, acceptType, principal);
}
UNREACHABLE();
@@ -2091,6 +2101,95 @@ Future<Response> Slave::Http::killNestedContainer(
});
}
+
+Future<Response> Slave::Http::attachContainerOutput(
+ const mesos::agent::Call& call,
+ ContentType contentType,
+ ContentType acceptType,
+ const Option<string>& principal) const
+{
+ CHECK_EQ(mesos::agent::Call::ATTACH_CONTAINER_OUTPUT, call.type());
+ CHECK(call.has_attach_container_output());
+
+ const ContainerID& containerId =
+ call.attach_container_output().container_id();
+
+ return slave->containerizer->attach(containerId)
+ .then([call, contentType, acceptType](Connection connection)
+ -> Future<Response> {
+ Request request;
+ request.method = "POST";
+ request.headers = {{"Accept", stringify(acceptType)},
+ {"Content-Type", stringify(contentType)}};
+
+ // The 'HOST' header must be EMPTY for non Internet addresses.
+ // TODO(vinod): Instead of setting domain to empty string (which results
+ // in an empty HOST header), add a new URL constructor that doesn't
+ // require domain or IP.
+ request.url.domain = "";
+
+ // NOTE: The path is currently ignored by the switch board.
+ request.url.path = "/";
+
+ request.type = Request::BODY;
+ request.body = serialize(contentType, call);
+
+ // We capture `connection` here to ensure that it doesn't go
+ // out of scope until the `onAny` handler on `transform` is executed.
+ return connection.send(request, true)
+ .then([connection, acceptType](const Response& response)
+ -> Future<Response> {
+ if (response.status != OK().status) {
+ return response;
+ }
+
+ // Evolve the `ProcessIO` records in the Response body to v1
+ // before sending them to the client.
+ Pipe pipe;
+ Pipe::Writer writer = pipe.writer();
+
+ OK ok;
+ ok.headers["Content-Type"] = stringify(acceptType);
+ ok.type = Response::PIPE;
+ ok.reader = pipe.reader();
+
+ CHECK_EQ(Response::PIPE, response.type);
+ CHECK_SOME(response.reader);
+ Pipe::Reader reader = response.reader.get();
+
+ auto deserializer = lambda::bind(
+ deserialize<ProcessIO>, acceptType, lambda::_1);
+
+ Owned<Reader<ProcessIO>> decoder(new Reader<ProcessIO>(
+ Decoder<ProcessIO>(deserializer), reader));
+
+ auto encoder = [acceptType](const ProcessIO& processIO) {
+ ::recordio::Encoder<v1::agent::ProcessIO> encoder (lambda::bind(
+ serialize, acceptType, lambda::_1));
+
+ return encoder.encode(evolve(processIO));
+ };
+
+ recordio::transform<ProcessIO>(std::move(decoder), encoder, writer)
+ .onAny([writer, reader, connection](
+ const Future<Nothing>& future) mutable {
+ CHECK(!future.isDiscarded());
+
+ if (future.isFailed()) {
+ writer.fail(future.failure());
+ reader.close();
+ return;
+ }
+
+ writer.close();
+ reader.close();
+ });
+
+ return ok;
+ });
+ });
+}
+
} // namespace slave {
} // namespace internal {
} // namespace mesos {
http://git-wip-us.apache.org/repos/asf/mesos/blob/d536a135/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index 0ede185..05865a0 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -645,6 +645,12 @@ private:
const Option<std::string>& principal,
ContentType contentType) const;
+ process::Future<process::http::Response> attachContainerOutput(
+ const mesos::agent::Call& call,
+ ContentType contentType,
+ ContentType acceptType,
+ const Option<std::string>& principal) const;
+
Slave* slave;
// Used to rate limit the statistics endpoint.
http://git-wip-us.apache.org/repos/asf/mesos/blob/d536a135/src/slave/validation.cpp
----------------------------------------------------------------------
diff --git a/src/slave/validation.cpp b/src/slave/validation.cpp
index b3d8bcd..46e8436 100644
--- a/src/slave/validation.cpp
+++ b/src/slave/validation.cpp
@@ -224,8 +224,23 @@ Option<Error> validate(
case mesos::agent::Call::LAUNCH_NESTED_CONTAINER_SESSION:
case mesos::agent::Call::ATTACH_CONTAINER_INPUT:
- case mesos::agent::Call::ATTACH_CONTAINER_OUTPUT:
return Error("Unsupported");
+
+ case mesos::agent::Call::ATTACH_CONTAINER_OUTPUT: {
+ if (!call.has_attach_container_output()) {
+ return Error("Expecting 'attach_container_output' to be present");
+ }
+
+ Option<Error> error = validation::container::validateContainerId(
+ call.attach_container_output().container_id());
+
+ if (error.isSome()) {
+ return Error("'attach_container_output.container_id' is invalid"
+ ": " + error->message);
+ }
+
+ return None();
+ }
}
UNREACHABLE();
http://git-wip-us.apache.org/repos/asf/mesos/blob/d536a135/src/tests/api_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/api_tests.cpp b/src/tests/api_tests.cpp
index 03e8590..b2a70f3 100644
--- a/src/tests/api_tests.cpp
+++ b/src/tests/api_tests.cpp
@@ -3616,6 +3616,43 @@ TEST_P(AgentAPITest, NestedContainerLaunch)
driver.join();
}
+
+// TODO(vinod): Update the test when mesos containerizer
+// adds support for `attach`.
+TEST_P(AgentAPITest, AttachContainerOutputFailure)
+{
+ ContentType contentType = GetParam();
+
+ Clock::pause();
+
+ Future<Nothing> __recover = FUTURE_DISPATCH(_, &Slave::__recover);
+
+ StandaloneMasterDetector detector;
+ Try<Owned<cluster::Slave>> slave = StartSlave(&detector);
+
+ ASSERT_SOME(slave);
+
+ // Wait for the agent to finish recovery.
+ AWAIT_READY(__recover);
+ Clock::settle();
+
+ v1::agent::Call call;
+ call.set_type(v1::agent::Call::ATTACH_CONTAINER_OUTPUT);
+
+ call.mutable_attach_container_output()->mutable_container_id()
+ ->set_value(UUID::random().toString());
+
+ Future<http::Response> response = http::post(
+ slave.get()->pid,
+ "api/v1",
+ createBasicAuthHeaders(DEFAULT_CREDENTIAL),
+ serialize(contentType, call),
+ stringify(contentType));
+
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::InternalServerError().status, response);
+ AWAIT_EXPECT_RESPONSE_BODY_EQ("Unsupported", response);
+}
+
} // namespace tests {
} // namespace internal {
} // namespace mesos {