You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@mesos.apache.org by an...@apache.org on 2016/12/01 06:43:34 UTC
[1/7] mesos git commit: Introduced streaming content types.
Repository: mesos
Updated Branches:
refs/heads/master 1c4c7c949 -> 48f6deb2d
Introduced streaming content types.
Added a couple of streaming content types that can be
used by handlers to support request streaming in the future.
Review: https://reviews.apache.org/r/54217/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/dfe21b93
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/dfe21b93
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/dfe21b93
Branch: refs/heads/master
Commit: dfe21b93784c2a44fa8696ad8733a7e3d23f1a17
Parents: 1c4c7c9
Author: Anand Mazumdar <an...@apache.org>
Authored: Wed Nov 30 22:41:30 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Wed Nov 30 22:41:30 2016 -0800
----------------------------------------------------------------------
include/mesos/http.hpp | 18 ++++++++++++++----
1 file changed, 14 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/dfe21b93/include/mesos/http.hpp
----------------------------------------------------------------------
diff --git a/include/mesos/http.hpp b/include/mesos/http.hpp
index bd7e81f..fdc4839 100644
--- a/include/mesos/http.hpp
+++ b/include/mesos/http.hpp
@@ -21,15 +21,25 @@
namespace mesos {
-const char APPLICATION_JSON[] = "application/json";
const char APPLICATION_PROTOBUF[] = "application/x-protobuf";
+const char APPLICATION_JSON[] = "application/json";
+
+
+// The content-types used for streaming requests.
+//
+// TODO(anand): These need to be revisited before calling
+// out any features that use them as _stable_.
+const char APPLICATION_STREAMING_PROTOBUF[] = "application/x-protobuf+recordio";
+const char APPLICATION_STREAMING_JSON[] = "application/json+recordio";
+
-// Possible content-types that can be used as responses for
-// the Mesos HTTP API.
+// Possible content-types that can be used for requests to the Mesos HTTP API.
enum class ContentType
{
PROTOBUF,
- JSON
+ JSON,
+ STREAMING_PROTOBUF,
+ STREAMING_JSON
};
[6/7] mesos git commit: Added a test for the `ATTACH_CONTAINER_INPUT`
call on the Agent API.
Posted by an...@apache.org.
Added a test for the `ATTACH_CONTAINER_INPUT` call on the Agent API.
Review: https://reviews.apache.org/r/54242/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/db4b832c
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/db4b832c
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/db4b832c
Branch: refs/heads/master
Commit: db4b832cbc66e1540a897b02d1cedb48c93c0722
Parents: 7e472c3
Author: Anand Mazumdar <an...@apache.org>
Authored: Wed Nov 30 22:42:05 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Wed Nov 30 22:42:05 2016 -0800
----------------------------------------------------------------------
src/tests/api_tests.cpp | 51 ++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 51 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/db4b832c/src/tests/api_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/api_tests.cpp b/src/tests/api_tests.cpp
index b2a70f3..a803408 100644
--- a/src/tests/api_tests.cpp
+++ b/src/tests/api_tests.cpp
@@ -3653,6 +3653,57 @@ TEST_P(AgentAPITest, AttachContainerOutputFailure)
AWAIT_EXPECT_RESPONSE_BODY_EQ("Unsupported", response);
}
+
+TEST_F(AgentAPITest, AttachContainerInputFailure)
+{
+ Clock::pause();
+
+ Future<Nothing> __recover = FUTURE_DISPATCH(_, &Slave::__recover);
+
+ StandaloneMasterDetector detector;
+ MockContainerizer mockContainerizer;
+
+ EXPECT_CALL(mockContainerizer, recover(_))
+ .WillOnce(Return(Future<Nothing>(Nothing())));
+
+ Try<Owned<cluster::Slave>> slave = StartSlave(&detector, &mockContainerizer);
+
+ 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_INPUT);
+
+ call.mutable_attach_container_input()->mutable_container_id()
+ ->set_value(UUID::random().toString());
+
+ ContentType contentType = ContentType::STREAMING_PROTOBUF;
+
+ ::recordio::Encoder<v1::agent::Call> encoder(lambda::bind(
+ serialize, contentType, lambda::_1));
+
+ EXPECT_CALL(mockContainerizer, attach(_))
+ .WillOnce(Return(process::Failure("Unsupported")));
+
+ Future<http::Response> response = http::post(
+ slave.get()->pid,
+ "api/v1",
+ createBasicAuthHeaders(DEFAULT_CREDENTIAL),
+ encoder.encode(call),
+ stringify(contentType));
+
+ AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::InternalServerError().status, response);
+ AWAIT_EXPECT_RESPONSE_BODY_EQ("Unsupported", response);
+
+ // The destructor of `cluster::Slave` will try to clean up any
+ // remaining containers by inspecting the result of `containers()`.
+ EXPECT_CALL(mockContainerizer, containers())
+ .WillRepeatedly(Return(hashset<ContainerID>()));
+}
+
} // namespace tests {
} // namespace internal {
} // namespace mesos {
[7/7] mesos git commit: Fixed the attach output container test to use
the mock containerizer.
Posted by an...@apache.org.
Fixed the attach output container test to use the mock containerizer.
Otherwise, the test would fail when we add support to the
containerizer to support `attach()`.
Review: https://reviews.apache.org/r/54243/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/48f6deb2
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/48f6deb2
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/48f6deb2
Branch: refs/heads/master
Commit: 48f6deb2dc21611b9f67fd1d906d786c95087dd2
Parents: db4b832
Author: Anand Mazumdar <an...@apache.org>
Authored: Wed Nov 30 22:42:10 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Wed Nov 30 22:42:10 2016 -0800
----------------------------------------------------------------------
src/tests/api_tests.cpp | 15 ++++++++++++++-
1 file changed, 14 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/48f6deb2/src/tests/api_tests.cpp
----------------------------------------------------------------------
diff --git a/src/tests/api_tests.cpp b/src/tests/api_tests.cpp
index a803408..ca4f1de 100644
--- a/src/tests/api_tests.cpp
+++ b/src/tests/api_tests.cpp
@@ -3628,7 +3628,12 @@ TEST_P(AgentAPITest, AttachContainerOutputFailure)
Future<Nothing> __recover = FUTURE_DISPATCH(_, &Slave::__recover);
StandaloneMasterDetector detector;
- Try<Owned<cluster::Slave>> slave = StartSlave(&detector);
+ MockContainerizer mockContainerizer;
+
+ EXPECT_CALL(mockContainerizer, recover(_))
+ .WillOnce(Return(Future<Nothing>(Nothing())));
+
+ Try<Owned<cluster::Slave>> slave = StartSlave(&detector, &mockContainerizer);
ASSERT_SOME(slave);
@@ -3642,6 +3647,9 @@ TEST_P(AgentAPITest, AttachContainerOutputFailure)
call.mutable_attach_container_output()->mutable_container_id()
->set_value(UUID::random().toString());
+ EXPECT_CALL(mockContainerizer, attach(_))
+ .WillOnce(Return(process::Failure("Unsupported")));
+
Future<http::Response> response = http::post(
slave.get()->pid,
"api/v1",
@@ -3651,6 +3659,11 @@ TEST_P(AgentAPITest, AttachContainerOutputFailure)
AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::InternalServerError().status, response);
AWAIT_EXPECT_RESPONSE_BODY_EQ("Unsupported", response);
+
+ // The destructor of `cluster::Slave` will try to clean up any
+ // remaining containers by inspecting the result of `containers()`.
+ EXPECT_CALL(mockContainerizer, containers())
+ .WillRepeatedly(Return(hashset<ContainerID>()));
}
[4/7] mesos git commit: Made the agent's `api/v1` handler handle
request streaming.
Posted by an...@apache.org.
Made the agent's `api/v1` handler handle request streaming.
This lays the ground work for adding the attach container input
call later.
Review: https://reviews.apache.org/r/54220/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/b11408de
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/b11408de
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/b11408de
Branch: refs/heads/master
Commit: b11408de0f2ead054cbdff66daade0952556064e
Parents: 15b2fd9
Author: Anand Mazumdar <an...@apache.org>
Authored: Wed Nov 30 22:41:49 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Wed Nov 30 22:41:49 2016 -0800
----------------------------------------------------------------------
src/common/http.cpp | 18 +++++++++++++
src/common/http.hpp | 4 +++
src/slave/http.cpp | 67 ++++++++++++++++++++++++++++++++++++++++--------
src/slave/slave.hpp | 7 ++++-
4 files changed, 84 insertions(+), 12 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/b11408de/src/common/http.cpp
----------------------------------------------------------------------
diff --git a/src/common/http.cpp b/src/common/http.cpp
index 8309e5e..05215ca 100644
--- a/src/common/http.cpp
+++ b/src/common/http.cpp
@@ -118,6 +118,24 @@ string serialize(
}
+bool requestStreaming(ContentType contentType)
+{
+ switch(contentType) {
+ case ContentType::PROTOBUF:
+ case ContentType::JSON: {
+ return false;
+ }
+
+ case ContentType::STREAMING_JSON:
+ case ContentType::STREAMING_PROTOBUF: {
+ return true;
+ }
+ }
+
+ UNREACHABLE();
+}
+
+
// TODO(bmahler): Kill these in favor of automatic Proto->JSON
// Conversion (when it becomes available).
http://git-wip-us.apache.org/repos/asf/mesos/blob/b11408de/src/common/http.hpp
----------------------------------------------------------------------
diff --git a/src/common/http.hpp b/src/common/http.hpp
index cb8eead..fc2ac14 100644
--- a/src/common/http.hpp
+++ b/src/common/http.hpp
@@ -90,6 +90,10 @@ Try<Message> deserialize(
}
+// Returns true if the content type can be used for streaming requests.
+bool requestStreaming(ContentType contentType);
+
+
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/b11408de/src/slave/http.cpp
----------------------------------------------------------------------
diff --git a/src/slave/http.cpp b/src/slave/http.cpp
index 87189dd..19afa75 100644
--- a/src/slave/http.cpp
+++ b/src/slave/http.cpp
@@ -330,10 +330,15 @@ Future<Response> Slave::Http::api(
contentType = ContentType::JSON;
} else if (contentType_.get() == APPLICATION_PROTOBUF) {
contentType = ContentType::PROTOBUF;
+ } else if (contentType_.get() == APPLICATION_STREAMING_JSON) {
+ contentType = ContentType::STREAMING_JSON;
+ } else if (contentType_.get() == APPLICATION_STREAMING_PROTOBUF) {
+ contentType = ContentType::STREAMING_PROTOBUF;
} else {
return UnsupportedMediaType(
string("Expecting 'Content-Type' of ") +
- APPLICATION_JSON + " or " + APPLICATION_PROTOBUF);
+ APPLICATION_JSON + " or " + APPLICATION_PROTOBUF + " or " +
+ APPLICATION_STREAMING_JSON + " or " + APPLICATION_STREAMING_PROTOBUF);
}
// This lambda deserializes a string into a valid `Call`
@@ -358,21 +363,48 @@ Future<Response> Slave::Http::api(
};
ContentType acceptType;
- if (request.acceptsMediaType(APPLICATION_JSON)) {
+ if (request.acceptsMediaType(APPLICATION_STREAMING_PROTOBUF)) {
+ acceptType = ContentType::STREAMING_PROTOBUF;
+ } else if (request.acceptsMediaType(APPLICATION_STREAMING_JSON)) {
+ acceptType = ContentType::STREAMING_JSON;
+ } else if (request.acceptsMediaType(APPLICATION_JSON)) {
acceptType = ContentType::JSON;
} else if (request.acceptsMediaType(APPLICATION_PROTOBUF)) {
acceptType = ContentType::PROTOBUF;
} else {
return NotAcceptable(
string("Expecting 'Accept' to allow ") +
- "'" + APPLICATION_PROTOBUF + "' or '" + APPLICATION_JSON + "'");
+ APPLICATION_JSON + " or " + APPLICATION_PROTOBUF + " or " +
+ APPLICATION_STREAMING_JSON + " or " + APPLICATION_STREAMING_PROTOBUF);
}
- if (contentType == ContentType::JSON ||
- contentType == ContentType::PROTOBUF) {
- CHECK_EQ(http::Request::PIPE, request.type);
- CHECK_SOME(request.reader);
+ CHECK_EQ(http::Request::PIPE, request.type);
+ CHECK_SOME(request.reader);
+
+ if (requestStreaming(contentType)) {
+ Owned<Reader<mesos::agent::Call>> reader(new Reader<mesos::agent::Call>(
+ Decoder<mesos::agent::Call>(lambda::bind(
+ deserializer, lambda::_1, contentType)), request.reader.get()));
+
+ return reader->read()
+ .then(defer(
+ slave->self(),
+ [=](const Result<mesos::agent::Call>& call) -> Future<Response> {
+ if (call.isNone()) {
+ return BadRequest("Received EOF while reading request body");
+ }
+
+ if (call.isError()) {
+ return Failure(call.error());
+ }
+ return _api(call.get(),
+ std::move(reader),
+ contentType,
+ acceptType,
+ principal);
+ }));
+ } else {
Pipe::Reader reader = request.reader.get(); // Remove const.
return reader.readAll()
@@ -383,21 +415,34 @@ Future<Response> Slave::Http::api(
if (call.isError()) {
return BadRequest(call.error());
}
- return _api(call.get(), contentType, acceptType, principal);
+ return _api(call.get(), None(), contentType, acceptType, principal);
}));
- } else {
- // TODO(vinod): Add support for 'streaming' content type.
- UNREACHABLE();
}
}
Future<Response> Slave::Http::_api(
const agent::Call& call,
+ Option<Owned<Reader<mesos::agent::Call>>>&& reader,
ContentType contentType,
ContentType acceptType,
const Option<string>& principal) const
{
+ // Validate that a client has not _accidentally_ sent us a
+ // streaming request for a call type that does not support it.
+ if (requestStreaming(contentType) &&
+ call.type() != mesos::agent::Call::ATTACH_CONTAINER_INPUT) {
+ return UnsupportedMediaType(
+ "Streaming 'Content-Type' " + stringify(contentType) + " is not "
+ "supported for " + stringify(call.type()) + " call");
+ } else if (!requestStreaming(contentType) &&
+ call.type() == mesos::agent::Call::ATTACH_CONTAINER_INPUT) {
+ return UnsupportedMediaType(
+ string("Expecting 'Content-Type' of ") + APPLICATION_STREAMING_JSON +
+ " or " + APPLICATION_STREAMING_PROTOBUF + " for " +
+ stringify(call.type()) + " call");
+ }
+
LOG(INFO) << "Processing call " << call.type();
switch (call.type()) {
http://git-wip-us.apache.org/repos/asf/mesos/blob/b11408de/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index 05865a0..070c8ae 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -63,6 +63,7 @@
#include "common/http.hpp"
#include "common/protobuf_utils.hpp"
+#include "common/recordio.hpp"
#include "files/files.hpp"
@@ -523,9 +524,13 @@ private:
private:
JSON::Object _flags() const;
- // Continuation for `/api` endpoint that handles non-streaming requests.
+ // Continuation for `/api` endpoint that handles streaming and non-streaming
+ // requests. In case of a streaming request, `call` would be the first
+ // record and additional records can be read using the `reader`. For
+ // non-streaming requests, `reader` would be set to `None()`.
process::Future<process::http::Response> _api(
const agent::Call& call,
+ Option<process::Owned<recordio::Reader<agent::Call>>>&& reader,
ContentType contentTye,
ContentType acceptType,
const Option<std::string>& principal) const;
[2/7] mesos git commit: Made `serialize()`/`deserialize()` handle
streaming content types.
Posted by an...@apache.org.
Made `serialize()`/`deserialize()` handle streaming content types.
The `serialize()` helper can now encode a record in 'Record-IO'
format for streaming content type. Also, `deserialize()` now
handles a streaming content type by just parsing it the body
into a corresponding protobuf.
Review: https://reviews.apache.org/r/54218/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/2c2abd70
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/2c2abd70
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/2c2abd70
Branch: refs/heads/master
Commit: 2c2abd70962a85c959b650b6fdd6177a106809d4
Parents: dfe21b9
Author: Anand Mazumdar <an...@apache.org>
Authored: Wed Nov 30 22:41:36 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Wed Nov 30 22:41:36 2016 -0800
----------------------------------------------------------------------
src/common/http.cpp | 13 +++++++++++--
src/common/http.hpp | 10 ++++++++--
2 files changed, 19 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/2c2abd70/src/common/http.cpp
----------------------------------------------------------------------
diff --git a/src/common/http.cpp b/src/common/http.cpp
index 42af330..8309e5e 100644
--- a/src/common/http.cpp
+++ b/src/common/http.cpp
@@ -37,6 +37,7 @@
#include <stout/duration.hpp>
#include <stout/foreach.hpp>
#include <stout/protobuf.hpp>
+#include <stout/recordio.hpp>
#include <stout/stringify.hpp>
#include <stout/unreachable.hpp>
@@ -72,6 +73,12 @@ ostream& operator<<(ostream& stream, ContentType contentType)
case ContentType::JSON: {
return stream << APPLICATION_JSON;
}
+ case ContentType::STREAMING_PROTOBUF: {
+ return stream << APPLICATION_STREAMING_PROTOBUF;
+ }
+ case ContentType::STREAMING_JSON: {
+ return stream << APPLICATION_STREAMING_JSON;
+ }
}
UNREACHABLE();
@@ -96,10 +103,12 @@ string serialize(
const google::protobuf::Message& message)
{
switch (contentType) {
- case ContentType::PROTOBUF: {
+ case ContentType::PROTOBUF:
+ case ContentType::STREAMING_PROTOBUF: {
return message.SerializeAsString();
}
- case ContentType::JSON: {
+ case ContentType::JSON:
+ case ContentType::STREAMING_JSON: {
JSON::Object object = JSON::protobuf(message);
return stringify(object);
}
http://git-wip-us.apache.org/repos/asf/mesos/blob/2c2abd70/src/common/http.hpp
----------------------------------------------------------------------
diff --git a/src/common/http.hpp b/src/common/http.hpp
index 378208b..cb8eead 100644
--- a/src/common/http.hpp
+++ b/src/common/http.hpp
@@ -50,6 +50,8 @@ extern hashset<std::string> AUTHORIZABLE_ENDPOINTS;
// Serializes a protobuf message for transmission
// based on the HTTP content type.
+// NOTE: For streaming `contentType`, `message` would not
+// be serialized in "Record-IO" format.
std::string serialize(
ContentType contentType,
const google::protobuf::Message& message);
@@ -57,20 +59,24 @@ std::string serialize(
// Deserializes a string message into a protobuf message based on the
// HTTP content type.
+// NOTE: For streaming `contentType`, `body` should not be
+// in "Record-IO" format.
template <typename Message>
Try<Message> deserialize(
ContentType contentType,
const std::string& body)
{
switch (contentType) {
- case ContentType::PROTOBUF: {
+ case ContentType::PROTOBUF:
+ case ContentType::STREAMING_PROTOBUF: {
Message message;
if (!message.ParseFromString(body)) {
return Error("Failed to parse body into a protobuf object");
}
return message;
}
- case ContentType::JSON: {
+ case ContentType::JSON:
+ case ContentType::STREAMING_JSON: {
Try<JSON::Value> value = JSON::parse(body);
if (value.isError()) {
return Error("Failed to parse body into JSON: " + value.error());
[5/7] mesos git commit: Added the v1 \`ATTACH_CONTAINER_INPUT\` call
handler on the agent.
Posted by an...@apache.org.
Added the v1 \`ATTACH_CONTAINER_INPUT\` call handler on the agent.
Review: https://reviews.apache.org/r/54221/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/7e472c37
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/7e472c37
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/7e472c37
Branch: refs/heads/master
Commit: 7e472c3763513571a1eb0db786fc02103a96d066
Parents: b11408d
Author: Anand Mazumdar <an...@apache.org>
Authored: Wed Nov 30 22:41:53 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Wed Nov 30 22:41:53 2016 -0800
----------------------------------------------------------------------
src/slave/http.cpp | 79 +++++++++++++++++++++++++++++++++++++++++++++++-
src/slave/slave.hpp | 7 +++++
2 files changed, 85 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/7e472c37/src/slave/http.cpp
----------------------------------------------------------------------
diff --git a/src/slave/http.cpp b/src/slave/http.cpp
index 19afa75..6915afe 100644
--- a/src/slave/http.cpp
+++ b/src/slave/http.cpp
@@ -498,9 +498,13 @@ Future<Response> Slave::Http::_api(
return killNestedContainer(call, principal, acceptType);
case mesos::agent::Call::LAUNCH_NESTED_CONTAINER_SESSION:
- case mesos::agent::Call::ATTACH_CONTAINER_INPUT:
return NotImplemented();
+ case mesos::agent::Call::ATTACH_CONTAINER_INPUT:
+ CHECK_SOME(reader);
+ return attachContainerInput(
+ call, std::move(reader).get(), contentType, acceptType, principal);
+
case mesos::agent::Call::ATTACH_CONTAINER_OUTPUT:
return attachContainerOutput(call, contentType, acceptType, principal);
}
@@ -2149,6 +2153,79 @@ Future<Response> Slave::Http::killNestedContainer(
}
+Future<Response> Slave::Http::attachContainerInput(
+ const mesos::agent::Call& call,
+ Owned<Reader<mesos::agent::Call>>&& decoder,
+ ContentType contentType,
+ ContentType acceptType,
+ const Option<string>& principal) const
+{
+ CHECK_EQ(mesos::agent::Call::ATTACH_CONTAINER_INPUT, call.type());
+ CHECK(call.has_attach_container_input());
+
+ const ContainerID& containerId = call.attach_container_input().container_id();
+
+ Pipe pipe;
+ Pipe::Reader reader = pipe.reader();
+ Pipe::Writer writer = pipe.writer();
+
+ auto encoder = [contentType](const mesos::agent::Call& call) {
+ ::recordio::Encoder<mesos::agent::Call> encoder(lambda::bind(
+ serialize, contentType, lambda::_1));
+
+ return encoder.encode(call);
+ };
+
+ // Write the first record. We had extracted it from the `decoder`
+ // in the `api()` handler to identify the call type earlier.
+ pipe.writer().write(encoder(call));
+
+ // We create this here since C++11 does not support move capture of `reader`.
+ Future<Nothing> transform = recordio::transform<mesos::agent::Call>(
+ std::move(decoder), encoder, writer);
+
+ return slave->containerizer->attach(containerId)
+ .then([contentType, acceptType, reader, writer, transform](
+ Connection connection) mutable {
+ Request request;
+ request.method = "POST";
+ request.type = Request::PIPE;
+ request.reader = reader;
+ request.headers = {{"Content-Type", stringify(contentType)},
+ {"Accept-Type", stringify(acceptType)}};
+
+ // See comments in `attachContainerOutput()` for the reasoning
+ // behind these values.
+ request.url.domain = "";
+ request.url.path = "/";
+
+ transform
+ .onAny([reader, writer](
+ const Future<Nothing>& future) mutable {
+ CHECK(!future.isDiscarded());
+
+ if (future.isFailed()) {
+ writer.fail(future.failure());
+ reader.close();
+ return;
+ }
+
+ writer.close();
+ reader.close();
+ });
+
+ // This is a non Keep-Alive request which means the connection
+ // will be closed when the response is received. Since the
+ // 'Connection' is reference-counted, we must maintain a copy
+ // until the disconnection occurs.
+ connection.disconnected()
+ .onAny([connection]() {});
+
+ return connection.send(request);
+ });
+}
+
+
Future<Response> Slave::Http::attachContainerOutput(
const mesos::agent::Call& call,
ContentType contentType,
http://git-wip-us.apache.org/repos/asf/mesos/blob/7e472c37/src/slave/slave.hpp
----------------------------------------------------------------------
diff --git a/src/slave/slave.hpp b/src/slave/slave.hpp
index 070c8ae..cfec9dd 100644
--- a/src/slave/slave.hpp
+++ b/src/slave/slave.hpp
@@ -650,6 +650,13 @@ private:
const Option<std::string>& principal,
ContentType contentType) const;
+ process::Future<process::http::Response> attachContainerInput(
+ const mesos::agent::Call& call,
+ process::Owned<recordio::Reader<agent::Call>>&& decoder,
+ ContentType contentType,
+ ContentType acceptType,
+ const Option<std::string>& principal) const;
+
process::Future<process::http::Response> attachContainerOutput(
const mesos::agent::Call& call,
ContentType contentType,
[3/7] mesos git commit: Added validation for the attach container
input call on the agent.
Posted by an...@apache.org.
Added validation for the attach container input call on the agent.
Review: https://reviews.apache.org/r/54219/
Project: http://git-wip-us.apache.org/repos/asf/mesos/repo
Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/15b2fd99
Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/15b2fd99
Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/15b2fd99
Branch: refs/heads/master
Commit: 15b2fd995e4eccd666a48c6ebca9c53f11210f53
Parents: 2c2abd7
Author: Anand Mazumdar <an...@apache.org>
Authored: Wed Nov 30 22:41:42 2016 -0800
Committer: Anand Mazumdar <an...@apache.org>
Committed: Wed Nov 30 22:41:42 2016 -0800
----------------------------------------------------------------------
src/slave/validation.cpp | 17 ++++++++++++++++-
1 file changed, 16 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/mesos/blob/15b2fd99/src/slave/validation.cpp
----------------------------------------------------------------------
diff --git a/src/slave/validation.cpp b/src/slave/validation.cpp
index 46e8436..4005cfc 100644
--- a/src/slave/validation.cpp
+++ b/src/slave/validation.cpp
@@ -223,9 +223,24 @@ Option<Error> validate(
}
case mesos::agent::Call::LAUNCH_NESTED_CONTAINER_SESSION:
- case mesos::agent::Call::ATTACH_CONTAINER_INPUT:
return Error("Unsupported");
+ case mesos::agent::Call::ATTACH_CONTAINER_INPUT: {
+ if (!call.has_attach_container_input()) {
+ return Error("Expecting 'attach_container_input' to be present");
+ }
+
+ Option<Error> error = validation::container::validateContainerId(
+ call.attach_container_input().container_id());
+
+ if (error.isSome()) {
+ return Error("'attach_container_input.container_id' is invalid"
+ ": " + error->message);
+ }
+
+ return None();
+ }
+
case mesos::agent::Call::ATTACH_CONTAINER_OUTPUT: {
if (!call.has_attach_container_output()) {
return Error("Expecting 'attach_container_output' to be present");