[mesos] 04/06: Improved performance of v1 agent operator API GET_TASKS call.
This is an automated email from the ASF dual-hosted git repository.
bmahler pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit 4aff0d7b583172ae890fbe944b268e222779956d
Author: Benjamin Mahler
AuthorDate: Thu Jan 30 15:40:59 2020 -0500
Improved performance of v1 agent operator API GET_TASKS call.
This follow the same approach used for the master's v1 calls:
https://github.com/apache/mesos/commit/d7dd4d0e8493331d7b7a21b504eb
https://github.com/apache/mesos/commit/3dda3622f5ed01e8c132dc5ca594
That is, serializing directly to protobuf or json from the in-memory
v0 state.
Review: https://reviews.apache.org/r/72066
---
src/slave/http.cpp | 341 +++--
src/slave/http.hpp | 4 +
2 files changed, 338 insertions(+), 7 deletions(-)
diff --git a/src/slave/http.cpp b/src/slave/http.cpp
index b120bf8..ab470cf 100644
--- a/src/slave/http.cpp
+++ b/src/slave/http.cpp
@@ -2143,7 +2143,7 @@ Future Http::getOperations(
Future Http::getTasks(
const mesos::agent::Call& call,
-ContentType acceptType,
+ContentType contentType,
const Option& principal) const
{
CHECK_EQ(mesos::agent::Call::GET_TASKS, call.type());
@@ -2156,19 +2156,346 @@ Future Http::getTasks(
{VIEW_FRAMEWORK, VIEW_TASK, VIEW_EXECUTOR})
.then(defer(
slave->self(),
-[this, acceptType](
+[this, contentType](
const Owned& approvers) -> Response {
- mesos::agent::Response response;
- response.set_type(mesos::agent::Response::GET_TASKS);
+ // Serialize the following message:
+ //
+ // v1::agent::Response response;
+ // response.set_type(mesos::agent::Response::GET_TASKS);
+ // *response.mutable_get_tasks() = _...;
- *response.mutable_get_tasks() = _getTasks(approvers);
+ switch (contentType) {
+case ContentType::PROTOBUF: {
+ string output;
+ google::protobuf::io::StringOutputStream stream();
+ google::protobuf::io::CodedOutputStream writer();
- return OK(serialize(acceptType, evolve(response)),
-stringify(acceptType));
+ WireFormatLite::WriteEnum(
+ v1::agent::Response::kTypeFieldNumber,
+ v1::agent::Response::GET_TASKS,
+ );
+
+ WireFormatLite::WriteBytes(
+ v1::agent::Response::kGetTasksFieldNumber,
+ serializeGetTasks(approvers),
+ );
+
+ // We must manually trim the unused buffer space since
+ // we use the string before the coded output stream is
+ // destructed.
+ writer.Trim();
+
+ return OK(std::move(output), stringify(contentType));
+}
+
+case ContentType::JSON: {
+ string body = jsonify([&](JSON::ObjectWriter* writer) {
+const google::protobuf::Descriptor* descriptor =
+ v1::agent::Response::descriptor();
+
+int field;
+
+field = v1::agent::Response::kTypeFieldNumber;
+writer->field(
+descriptor->FindFieldByNumber(field)->name(),
+v1::agent::Response::Type_Name(
+v1::agent::Response::GET_TASKS));
+
+field = v1::agent::Response::kGetTasksFieldNumber;
+writer->field(
+descriptor->FindFieldByNumber(field)->name(),
+jsonifyGetTasks(approvers));
+ });
+
+ // TODO(bmahler): Pass jsonp query parameter through here.
+ return OK(std::move(body), stringify(contentType));
+}
+
+default:
+ return NotAcceptable("Request must accept json or protobuf");
+ }
}));
}
+function Http::jsonifyGetTasks(
+const Owned& approvers) const
+{
+ return [=](JSON::ObjectWriter* writer) {
+// Construct framework list with both active and completed frameworks.
+vector frameworks;
+foreachvalue (Framework* f, slave->frameworks) {
+ if (approvers->approved(f->info)) {
+frameworks.push_back(f);
+ }
+}
+foreachvalue (const Owned& f, slave->completedFrameworks) {
+ if (approvers->approved(f->info)) {
+frameworks.push_back(f.get());
+ }
+}
+
+// Construct executor list with both active and completed executors.
+hashmap executors;
+foreach (const Framework* f, frameworks) {
+ foreachvalue (const Executor* e, f->executors) {
+if (approvers->approved(e->info, f->info)) {
+ executors.put(e, f);
+}
+ }
+ foreach (const Owned& e, f->completedExecutors) {
+if (approvers->approved(e->info, f->info)) {
+ executors.put(e.get(), f);
+}
+
[mesos] 06/06: Minor cleanups to master's GET_* call handlers.
This is an automated email from the ASF dual-hosted git repository.
bmahler pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit de8875498a799af20ff15eda68064c644dcc6d66
Author: Benjamin Mahler
AuthorDate: Mon Feb 24 21:04:43 2020 -0500
Minor cleanups to master's GET_* call handlers.
These cleanups are based on the more recent changes to the agent's
GET_* handlers, where these were noticed in code review for the
agent code but the code on the master side has these same issues.
---
src/master/readonly_handler.cpp | 23 ++-
1 file changed, 10 insertions(+), 13 deletions(-)
diff --git a/src/master/readonly_handler.cpp b/src/master/readonly_handler.cpp
index 341a75a..f9c0006 100644
--- a/src/master/readonly_handler.cpp
+++ b/src/master/readonly_handler.cpp
@@ -1218,7 +1218,7 @@ pair>
// Construct framework list with both active and completed frameworks.
vector frameworks;
- foreachvalue (Framework* framework, master->frameworks.registered) {
+ foreachvalue (const Framework* framework, master->frameworks.registered) {
// Skip unauthorized frameworks or frameworks without matching
// framework ID.
if (!selectFrameworkId.accept(framework->id()) ||
@@ -1246,7 +1246,6 @@ pair>
vector tasks;
foreach (const Framework* framework, frameworks) {
foreachvalue (Task* task, framework->tasks) {
- CHECK_NOTNULL(task);
// Skip unauthorized tasks or tasks without matching task ID.
if (!selectTaskId.accept(task->task_id()) ||
!approvers->approved(*task, framework->info)) {
@@ -1715,7 +1714,7 @@ function
return [=](JSON::ObjectWriter* writer) {
// Construct framework list with both active and completed frameworks.
vector frameworks;
-foreachvalue (Framework* framework, master->frameworks.registered) {
+foreachvalue (const Framework* framework, master->frameworks.registered) {
// Skip unauthorized frameworks.
if (approvers->approved(framework->info)) {
frameworks.push_back(framework);
@@ -1785,7 +1784,7 @@ string Master::ReadOnlyHandler::serializeGetExecutors(
{
// Construct framework list with both active and completed frameworks.
vector frameworks;
- foreachvalue (Framework* framework, master->frameworks.registered) {
+ foreachvalue (const Framework* framework, master->frameworks.registered) {
// Skip unauthorized frameworks.
if (approvers->approved(framework->info)) {
frameworks.push_back(framework);
@@ -1965,7 +1964,7 @@ function
Master::ReadOnlyHandler::jsonifyGetTasks(
return [=](JSON::ObjectWriter* writer) {
// Construct framework list with both active and completed frameworks.
vector frameworks;
-foreachvalue (Framework* framework, master->frameworks.registered) {
+foreachvalue (const Framework* framework, master->frameworks.registered) {
// Skip unauthorized frameworks.
if (approvers->approved(framework->info)) {
frameworks.push_back(framework);
@@ -2010,8 +2009,7 @@ function
Master::ReadOnlyHandler::jsonifyGetTasks(
descriptor->FindFieldByNumber(field)->name(),
[&](JSON::ArrayWriter* writer) {
foreach (const Framework* framework, frameworks) {
-foreachvalue (Task* task, framework->tasks) {
- CHECK_NOTNULL(task);
+foreachvalue (const Task* task, framework->tasks) {
// Skip unauthorized tasks.
if (!approvers->approved(*task, framework->info)) {
continue;
@@ -2065,7 +2063,7 @@ string Master::ReadOnlyHandler::serializeGetTasks(
{
// Construct framework list with both active and completed frameworks.
vector frameworks;
- foreachvalue (Framework* framework, master->frameworks.registered) {
+ foreachvalue (const Framework* framework, master->frameworks.registered) {
// Skip unauthorized frameworks.
if (approvers->approved(framework->info)) {
frameworks.push_back(framework);
@@ -2118,8 +2116,7 @@ string Master::ReadOnlyHandler::serializeGetTasks(
}
// Active tasks.
-foreachvalue (Task* task, framework->tasks) {
- CHECK_NOTNULL(task);
+foreachvalue (const Task* task, framework->tasks) {
// Skip unauthorized tasks.
if (!approvers->approved(*task, framework->info)) {
continue;
@@ -2276,16 +2273,16 @@ pair>
response.mutable_get_operations();
foreachvalue (const Slave* slave, master->slaves.registered) {
-foreachvalue (Operation* operation, slave->operations) {
+foreachvalue (const Operation* operation, slave->operations) {
if (approved(*operation)) {
operations->add_operations()->CopyFrom(*operation);
}
}
foreachvalue (
-const Slave::ResourceProvider resourceProvider,
+const Slave::ResourceProvider& resourceProvider,
slave->resourceProviders) {
- foreachvalue (Operation* operation,
[mesos] branch master updated (ea670ec -> de88754)
This is an automated email from the ASF dual-hosted git repository. bmahler pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/mesos.git. from ea670ec Fixed systemd socket activation on old systemd versions. new df9e3a6 Improved performance of v1 agent operator API GET_METRICS call. new ea2dcb4 Improved performance of v1 agent operator API GET_FRAMEWORKS call. new fd2c783 Improved performance of v1 agent operator API GET_EXECUTORS call. new 4aff0d7 Improved performance of v1 agent operator API GET_TASKS call. new 0d8c492 Improved performance of v1 agent operator API GET_STATE call. new de88754 Minor cleanups to master's GET_* call handlers. The 6 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: src/common/http.hpp | 110 + src/master/http.cpp | 108 + src/master/readonly_handler.cpp | 23 +- src/slave/http.cpp | 982 ++-- src/slave/http.hpp | 16 + 5 files changed, 1080 insertions(+), 159 deletions(-)
[mesos] 01/06: Improved performance of v1 agent operator API GET_METRICS call.
This is an automated email from the ASF dual-hosted git repository.
bmahler pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit df9e3a655467a3036ac7540c13266b6725654572
Author: Benjamin Mahler
AuthorDate: Tue Jan 28 13:42:36 2020 -0500
Improved performance of v1 agent operator API GET_METRICS call.
This follow the same approach used for the master's GET_METRICS call:
https://github.com/apache/mesos/commit/469f2ebaf65b1642d1eb4a1df81a
https://github.com/apache/mesos/commit/3dda3622f5ed01e8c132dc5ca594
That is, serializing directly to protobuf or json from the in-memory
v0 state.
Review: https://reviews.apache.org/r/72056
---
src/common/http.hpp | 110
src/master/http.cpp | 108 ++-
src/slave/http.cpp | 76 +---
3 files changed, 176 insertions(+), 118 deletions(-)
diff --git a/src/common/http.hpp b/src/common/http.hpp
index 5fc19fd..4a0f4a8 100644
--- a/src/common/http.hpp
+++ b/src/common/http.hpp
@@ -214,6 +214,116 @@ JSON::Object model(const FileInfo& fileInfo);
void json(JSON::ObjectWriter* writer, const Task& task);
+
+// NOTE: The `metrics` object provided as an argument must outlive
+// the returned function, since the function captures `metrics`
+// by reference to avoid a really expensive map copy. This is a
+// rather unsafe approach, but in typical jsonify usage this is
+// not an issue.
+//
+// TODO(bmahler): Use std::enable_if with std::is_same to check
+// that T is either the master or agent GetMetrics.
+template
+std::function jsonifyGetMetrics(
+const std::map& metrics)
+{
+ // Serialize the following message:
+ //
+ // mesos::master::Response::GetMetrics getMetrics;
+ // // or: mesos::agent::Response::GetMetrics getMetrics;
+ //
+ // foreachpair (const string& key, double value, metrics) {
+ // Metric* metric = getMetrics->add_metrics();
+ // metric->set_name(key);
+ // metric->set_value(value);
+ // }
+
+ return [&](JSON::ObjectWriter* writer) {
+const google::protobuf::Descriptor* descriptor = T::descriptor();
+
+int field = T::kMetricsFieldNumber;
+
+writer->field(
+descriptor->FindFieldByNumber(field)->name(),
+[&](JSON::ArrayWriter* writer) {
+ foreachpair (const std::string& key, double value, metrics) {
+writer->element([&](JSON::ObjectWriter* writer) {
+ const google::protobuf::Descriptor* descriptor =
+v1::Metric::descriptor();
+
+ int field;
+
+ field = v1::Metric::kNameFieldNumber;
+ writer->field(
+ descriptor->FindFieldByNumber(field)->name(), key);
+
+ field = v1::Metric::kValueFieldNumber;
+ writer->field(
+ descriptor->FindFieldByNumber(field)->name(), value);
+});
+ }
+});
+ };
+}
+
+
+// TODO(bmahler): Use std::enable_if with std::is_same to check
+// that T is either the master or agent GetMetrics.
+template
+std::string serializeGetMetrics(
+const std::map& metrics)
+{
+ // Serialize the following message:
+ //
+ // v1::master::Response::GetMetrics getMetrics;
+ // // or: v1::agent::Response::GetMetrics getMetrics;
+ //
+ // foreachpair (const string& key, double value, metrics) {
+ // Metric* metric = getMetrics->add_metrics();
+ // metric->set_name(key);
+ // metric->set_value(value);
+ // }
+
+ auto serializeMetric = [](const std::string& key, double value) {
+std::string output;
+google::protobuf::io::StringOutputStream stream();
+google::protobuf::io::CodedOutputStream writer();
+
+google::protobuf::internal::WireFormatLite::WriteString(
+v1::Metric::kNameFieldNumber, key, );
+google::protobuf::internal::WireFormatLite::WriteDouble(
+v1::Metric::kValueFieldNumber, value, );
+
+// While an explicit Trim() isn't necessary (since the coded
+// output stream is destructed before the string is returned),
+// it's a quite tricky bug to diagnose if Trim() is missed, so
+// we always do it explicitly to signal the reader about this
+// subtlety.
+writer.Trim();
+return output;
+ };
+
+ std::string output;
+ google::protobuf::io::StringOutputStream stream();
+ google::protobuf::io::CodedOutputStream writer();
+
+ foreachpair (const std::string& key, double value, metrics) {
+google::protobuf::internal::WireFormatLite::WriteBytes(
+T::kMetricsFieldNumber,
+serializeMetric(key, value),
+);
+ }
+
+ // While an explicit Trim() isn't necessary (since the coded
+ // output stream is destructed before the string is returned),
+ // it's a quite tricky bug to diagnose if Trim() is missed, so
+ // we always do it explicitly to signal the reader about this
+ // subtlety.
+
[mesos] 05/06: Improved performance of v1 agent operator API GET_STATE call.
This is an automated email from the ASF dual-hosted git repository.
bmahler pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit 0d8c492f7560aecf78d2c9f9ff4dc5f9c980e17b
Author: Benjamin Mahler
AuthorDate: Thu Jan 30 15:43:15 2020 -0500
Improved performance of v1 agent operator API GET_STATE call.
This follow the same approach used for the master's v1 calls:
https://github.com/apache/mesos/commit/1c60f0e4acbac96c34bd90e26515
https://github.com/apache/mesos/commit/3dda3622f5ed01e8c132dc5ca594
That is, serializing directly to protobuf or json from the in-memory
v0 state.
Review: https://reviews.apache.org/r/72067
---
src/slave/http.cpp | 139 ++---
src/slave/http.hpp | 4 ++
2 files changed, 136 insertions(+), 7 deletions(-)
diff --git a/src/slave/http.cpp b/src/slave/http.cpp
index ab470cf..d598440 100644
--- a/src/slave/http.cpp
+++ b/src/slave/http.cpp
@@ -2690,7 +2690,7 @@ Future Http::getResourceProviders(
Future Http::getState(
const mesos::agent::Call& call,
-ContentType acceptType,
+ContentType contentType,
const Option& principal) const
{
CHECK_EQ(mesos::agent::Call::GET_STATE, call.type());
@@ -2704,13 +2704,138 @@ Future Http::getState(
.then(defer(
slave->self(),
[=](const Owned& approvers) -> Response {
- mesos::agent::Response response;
- response.set_type(mesos::agent::Response::GET_STATE);
- *response.mutable_get_state() = _getState(approvers);
+ // Serialize the following message:
+ //
+ // v1::agent::Response response;
+ // response.set_type(mesos::agent::Response::GET_STATE);
+ // *response.mutable_get_state() = _...;
- return OK(serialize(acceptType, evolve(response)),
-stringify(acceptType));
-}));
+ switch (contentType) {
+case ContentType::PROTOBUF: {
+ string output;
+ google::protobuf::io::StringOutputStream stream();
+ google::protobuf::io::CodedOutputStream writer();
+
+ WireFormatLite::WriteEnum(
+ v1::agent::Response::kTypeFieldNumber,
+ v1::agent::Response::GET_STATE,
+ );
+
+ WireFormatLite::WriteBytes(
+ v1::agent::Response::kGetStateFieldNumber,
+ serializeGetState(approvers),
+ );
+
+ // We must manually trim the unused buffer space since
+ // we use the string before the coded output stream is
+ // destructed.
+ writer.Trim();
+
+ return OK(std::move(output), stringify(contentType));
+}
+
+case ContentType::JSON: {
+ string body = jsonify([&](JSON::ObjectWriter* writer) {
+const google::protobuf::Descriptor* descriptor =
+ v1::agent::Response::descriptor();
+
+int field;
+
+field = v1::agent::Response::kTypeFieldNumber;
+writer->field(
+descriptor->FindFieldByNumber(field)->name(),
+v1::agent::Response::Type_Name(
+v1::agent::Response::GET_STATE));
+
+field = v1::agent::Response::kGetStateFieldNumber;
+writer->field(
+descriptor->FindFieldByNumber(field)->name(),
+jsonifyGetState(approvers));
+ });
+
+ // TODO(bmahler): Pass jsonp query parameter through here.
+ return OK(std::move(body), stringify(contentType));
+}
+
+default:
+ return NotAcceptable("Request must accept json or protobuf");
+ }
+ }));
+}
+
+
+function Http::jsonifyGetState(
+const Owned& approvers) const
+{
+ // Serialize the following message:
+ //
+ // v1::agent::Response::GetState getState;
+ // *getState.mutable_get_tasks() = ...;
+ // *getState.mutable_get_executors() = ...;
+ // *getState.mutable_get_frameworks() = ...;
+
+ // TODO(bmahler): This copies the Owned object approvers.
+ return [=](JSON::ObjectWriter* writer) {
+const google::protobuf::Descriptor* descriptor =
+ v1::agent::Response::GetState::descriptor();
+
+int field;
+
+field = v1::agent::Response::GetState::kGetTasksFieldNumber;
+writer->field(
+descriptor->FindFieldByNumber(field)->name(),
+jsonifyGetTasks(approvers));
+
+field = v1::agent::Response::GetState::kGetExecutorsFieldNumber;
+writer->field(
+descriptor->FindFieldByNumber(field)->name(),
+jsonifyGetExecutors(approvers));
+
+field = v1::agent::Response::GetState::kGetFrameworksFieldNumber;
+writer->field(
+descriptor->FindFieldByNumber(field)->name(),
+
[mesos] 02/06: Improved performance of v1 agent operator API GET_FRAMEWORKS call.
This is an automated email from the ASF dual-hosted git repository.
bmahler pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit ea2dcb4aed02b7406c562b37e3d0afaeda0304b6
Author: Benjamin Mahler
AuthorDate: Thu Jan 30 15:37:55 2020 -0500
Improved performance of v1 agent operator API GET_FRAMEWORKS call.
This follow the same approach used for the master's v1 calls:
https://github.com/apache/mesos/commit/4f4dab961bd45ca444d13b831cdb
https://github.com/apache/mesos/commit/3dda3622f5ed01e8c132dc5ca594
That is, serializing directly to protobuf or json from the in-memory
v0 state.
Review: https://reviews.apache.org/r/72064
---
src/slave/http.cpp | 202 +++--
src/slave/http.hpp | 4 ++
2 files changed, 199 insertions(+), 7 deletions(-)
diff --git a/src/slave/http.cpp b/src/slave/http.cpp
index 095e787..bc8f222 100644
--- a/src/slave/http.cpp
+++ b/src/slave/http.cpp
@@ -1555,7 +1555,7 @@ Future Http::state(
Future Http::getFrameworks(
const mesos::agent::Call& call,
-ContentType acceptType,
+ContentType contentType,
const Option& principal) const
{
CHECK_EQ(mesos::agent::Call::GET_FRAMEWORKS, call.type());
@@ -1565,18 +1565,206 @@ Future Http::getFrameworks(
return ObjectApprovers::create(slave->authorizer, principal,
{VIEW_FRAMEWORK})
.then(defer(
slave->self(),
-[this, acceptType](
+[this, contentType](
const Owned& approvers) -> Response {
- mesos::agent::Response response;
- response.set_type(mesos::agent::Response::GET_FRAMEWORKS);
- *response.mutable_get_frameworks() = _getFrameworks(approvers);
+ // Serialize the following message:
+ //
+ // v1::agent::Response response;
+ // response.set_type(mesos::agent::Response::GET_FRAMEWORKS);
+ // *response.mutable_get_frameworks() = _...;
+
+ switch (contentType) {
+case ContentType::PROTOBUF: {
+ string output;
+ google::protobuf::io::StringOutputStream stream();
+ google::protobuf::io::CodedOutputStream writer();
+
+ WireFormatLite::WriteEnum(
+ v1::agent::Response::kTypeFieldNumber,
+ v1::agent::Response::GET_FRAMEWORKS,
+ );
+
+ WireFormatLite::WriteBytes(
+ v1::agent::Response::kGetFrameworksFieldNumber,
+ serializeGetFrameworks(approvers),
+ );
+
+ // We must manually trim the unused buffer space since
+ // we use the string before the coded output stream is
+ // destructed.
+ writer.Trim();
+
+ return OK(std::move(output), stringify(contentType));
+}
- return OK(serialize(acceptType, evolve(response)),
-stringify(acceptType));
+case ContentType::JSON: {
+ string body = jsonify([&](JSON::ObjectWriter* writer) {
+const google::protobuf::Descriptor* descriptor =
+ v1::agent::Response::descriptor();
+
+int field;
+
+field = v1::agent::Response::kTypeFieldNumber;
+writer->field(
+descriptor->FindFieldByNumber(field)->name(),
+v1::agent::Response::Type_Name(
+v1::agent::Response::GET_FRAMEWORKS));
+
+field = v1::agent::Response::kGetFrameworksFieldNumber;
+writer->field(
+descriptor->FindFieldByNumber(field)->name(),
+jsonifyGetFrameworks(approvers));
+ });
+
+ // TODO(bmahler): Pass jsonp query parameter through here.
+ return OK(std::move(body), stringify(contentType));
+}
+
+default:
+ return NotAcceptable("Request must accept json or protobuf");
+ }
}));
}
+function Http::jsonifyGetFrameworks(
+const Owned& approvers) const
+{
+ // Serialize the following message:
+ //
+ // v1::agent::Response::GetFrameworks getFrameworks;
+ //
+ // for each framework:
+ // *getFrameworks.add_frameworks()
+ // ->mutable_framework_info() = ...;
+ //
+ // for each completed framework:
+ // *getFrameworks.add_completed_frameworks()
+ // ->mutable_framework_info() = ...;
+
+ // Lambda for jsonifying the following message:
+ //
+ // v1::agent::Response::GetFrameworks::Framework framework;
+ // *framework.mutable_framework_info() = frameworkInfo;
+ auto jsonifyGetFramework = [](const FrameworkInfo& f) {
+return [&](JSON::ObjectWriter* writer) {
+ const google::protobuf::Descriptor* descriptor =
+v1::agent::Response::GetFrameworks::Framework::descriptor();
+
+ int field =
[mesos] 03/06: Improved performance of v1 agent operator API GET_EXECUTORS call.
This is an automated email from the ASF dual-hosted git repository.
bmahler pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit fd2c7837f52a5148959f006591dd91aa6e83ae7e
Author: Benjamin Mahler
AuthorDate: Thu Jan 30 15:39:36 2020 -0500
Improved performance of v1 agent operator API GET_EXECUTORS call.
This follow the same approach used for the master's v1 calls:
https://github.com/apache/mesos/commit/6ab835459a452e53fec8982a5aaa
https://github.com/apache/mesos/commit/3dda3622f5ed01e8c132dc5ca594
That is, serializing directly to protobuf or json from the in-memory
v0 state.
Review: https://reviews.apache.org/r/72065
---
src/slave/http.cpp | 224 +++--
src/slave/http.hpp | 4 +
2 files changed, 221 insertions(+), 7 deletions(-)
diff --git a/src/slave/http.cpp b/src/slave/http.cpp
index bc8f222..b120bf8 100644
--- a/src/slave/http.cpp
+++ b/src/slave/http.cpp
@@ -1795,7 +1795,7 @@ mesos::agent::Response::GetFrameworks
Http::_getFrameworks(
Future Http::getExecutors(
const mesos::agent::Call& call,
-ContentType acceptType,
+ContentType contentType,
const Option& principal) const
{
CHECK_EQ(mesos::agent::Call::GET_EXECUTORS, call.type());
@@ -1808,19 +1808,229 @@ Future Http::getExecutors(
{VIEW_FRAMEWORK, VIEW_EXECUTOR})
.then(defer(
slave->self(),
-[this, acceptType](
+[this, contentType](
const Owned& approvers) -> Response {
- mesos::agent::Response response;
- response.set_type(mesos::agent::Response::GET_EXECUTORS);
+ // Serialize the following message:
+ //
+ // v1::agent::Response response;
+ // response.set_type(mesos::agent::Response::GET_EXECUTORS);
+ // *response.mutable_get_executors() = _...;
- *response.mutable_get_executors() = _getExecutors(approvers);
+ switch (contentType) {
+case ContentType::PROTOBUF: {
+ string output;
+ google::protobuf::io::StringOutputStream stream();
+ google::protobuf::io::CodedOutputStream writer();
- return OK(serialize(acceptType, evolve(response)),
-stringify(acceptType));
+ WireFormatLite::WriteEnum(
+ v1::agent::Response::kTypeFieldNumber,
+ v1::agent::Response::GET_EXECUTORS,
+ );
+
+ WireFormatLite::WriteBytes(
+ v1::agent::Response::kGetExecutorsFieldNumber,
+ serializeGetExecutors(approvers),
+ );
+
+ // We must manually trim the unused buffer space since
+ // we use the string before the coded output stream is
+ // destructed.
+ writer.Trim();
+
+ return OK(std::move(output), stringify(contentType));
+}
+
+case ContentType::JSON: {
+ string body = jsonify([&](JSON::ObjectWriter* writer) {
+const google::protobuf::Descriptor* descriptor =
+ v1::agent::Response::descriptor();
+
+int field;
+
+field = v1::agent::Response::kTypeFieldNumber;
+writer->field(
+descriptor->FindFieldByNumber(field)->name(),
+v1::agent::Response::Type_Name(
+v1::agent::Response::GET_EXECUTORS));
+
+field = v1::agent::Response::kGetExecutorsFieldNumber;
+writer->field(
+descriptor->FindFieldByNumber(field)->name(),
+jsonifyGetExecutors(approvers));
+ });
+
+ // TODO(bmahler): Pass jsonp query parameter through here.
+ return OK(std::move(body), stringify(contentType));
+}
+
+default:
+ return NotAcceptable("Request must accept json or protobuf");
+ }
}));
}
+function Http::jsonifyGetExecutors(
+const Owned& approvers) const
+{
+ return [=](JSON::ObjectWriter* writer) {
+// Construct framework list with both active and completed frameworks.
+vector frameworks;
+foreachvalue (const Framework* f, slave->frameworks) {
+ if (approvers->approved(f->info)) {
+frameworks.push_back(f);
+ }
+}
+foreachvalue (const Owned& f, slave->completedFrameworks) {
+ if (approvers->approved(f->info)) {
+frameworks.push_back(f.get());
+ }
+}
+
+// Lambda for jsonifying the following message:
+//
+// v1::agent::Response::GetExecutors::Executor executor;
+// *executor.mutable_executor_info() = executorInfo;
+auto jsonifyGetExecutor = [](const ExecutorInfo& e) {
+ return [&](JSON::ObjectWriter* writer) {
+const google::protobuf::Descriptor* descriptor =
+
[mesos] branch master updated (cb8106f -> ea670ec)
This is an automated email from the ASF dual-hosted git repository. abudnik pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/mesos.git. from cb8106f Made `http`, `pid` and `heartbeater` of `Framework` private. new 360e883 Removed reimplementation of `cloexec` from systemd activation code. new ea670ec Fixed systemd socket activation on old systemd versions. The 2 revisions listed above as "new" are entirely new to this repository and will be described in separate emails. The revisions listed as "add" were already present in the repository and have only been added to this reference. Summary of changes: src/linux/systemd.cpp | 32 src/linux/systemd.hpp | 3 ++- src/slave/main.cpp| 5 - 3 files changed, 10 insertions(+), 30 deletions(-)
[mesos] 02/02: Fixed systemd socket activation on old systemd versions.
This is an automated email from the ASF dual-hosted git repository.
abudnik pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit ea670eca55af847bd7fcba1510726dd0059edf32
Author: Andrei Budnik
AuthorDate: Thu Feb 20 17:30:18 2020 +0100
Fixed systemd socket activation on old systemd versions.
This patch fixes the bug when `listenFdsWithName` function returns
an empty set of file descriptors on pre-227 systemd versions when
`domain_socket_location` value is not equals to the "systemd:unknown".
This happens when a user expects a newer version of systemd and
specifies a "systemd/", but
the actual systemd version does not support `FileDescriptorName`.
In this case, `LISTEN_FDNAMES` env variable is not specified,
so all socket FDs, which are specified by the `LISTEN_FDS`,
must be used to locate the path to the domain socket.
Review: https://reviews.apache.org/r/72158
---
src/linux/systemd.cpp | 6 +++---
src/linux/systemd.hpp | 3 ++-
src/slave/main.cpp| 5 -
3 files changed, 9 insertions(+), 5 deletions(-)
diff --git a/src/linux/systemd.cpp b/src/linux/systemd.cpp
index c7815a0..75d3c2b 100644
--- a/src/linux/systemd.cpp
+++ b/src/linux/systemd.cpp
@@ -395,7 +395,7 @@ Try> listenFds()
}
-Try> listenFdsWithName(const std::string& name)
+Try> listenFdsWithNames(const hashset& names)
{
Try> fds = listenFds();
if (fds.isError()) {
@@ -420,8 +420,8 @@ Try> listenFdsWithName(const std::string&
name)
}
std::vector result;
- for (size_t i=0; i < listenFdnames.size(); ++i) {
-if (listenFdnames[i] == name) {
+ for (size_t i = 0; i < listenFdnames.size(); ++i) {
+if (names.contains(listenFdnames[i])) {
result.push_back(fds->at(i));
}
}
diff --git a/src/linux/systemd.hpp b/src/linux/systemd.hpp
index ba96000..e3644c0 100644
--- a/src/linux/systemd.hpp
+++ b/src/linux/systemd.hpp
@@ -20,6 +20,7 @@
#include
#include
+#include
#include
#include
#include
@@ -101,7 +102,7 @@ Try> listenFds();
// The names are set by the `FileDescriptorName=` directive in the unit file.
// This requires systemd 227 or newer. Since any number of unit files can
// specify the same name, this can return more than one file descriptor.
-Try> listenFdsWithName(const std::string& name);
+Try> listenFdsWithNames(const hashset& names);
// Clear the `$LISTEN_PID`, `$LISTEN_FDS` and `$LISTEN_FDNAMES` environment
// variables.
diff --git a/src/slave/main.cpp b/src/slave/main.cpp
index c1e6551..0aa2cc9 100644
--- a/src/slave/main.cpp
+++ b/src/slave/main.cpp
@@ -651,8 +651,11 @@ int main(int argc, char** argv)
// Chop off `systemd:` prefix.
std::string name = flags.domain_socket_location->substr(8);
#ifdef __linux__
+ // NOTE: Some systemd versions do not support FileDescriptorName,
+ // thus we also need to listen on descriptors with name "unknown",
+ // which is used for sockets where no name could be determined.
Try> socketFds =
-systemd::socket_activation::listenFdsWithName(name);
+systemd::socket_activation::listenFdsWithNames({name, "unknown"});
#else
Try> socketFds =
Try>({}); // Dummy to avoid compile errors.
[mesos] 01/02: Removed reimplementation of `cloexec` from systemd activation code.
This is an automated email from the ASF dual-hosted git repository.
abudnik pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/mesos.git
commit 360e8838d08eb07496b2c3e6bcd511348ce488b0
Author: Andrei Budnik
AuthorDate: Fri Feb 21 12:08:47 2020 +0100
Removed reimplementation of `cloexec` from systemd activation code.
Review: https://reviews.apache.org/r/72157
---
src/linux/systemd.cpp | 26 +-
1 file changed, 1 insertion(+), 25 deletions(-)
diff --git a/src/linux/systemd.cpp b/src/linux/systemd.cpp
index 9897473..c7815a0 100644
--- a/src/linux/systemd.cpp
+++ b/src/linux/systemd.cpp
@@ -337,30 +337,6 @@ Try start(const string& name)
namespace socket_activation {
-static Try setCloexecFlag(int fd, bool cloexec)
-{
- CHECK(fd >= 0) << "Invalid file desciptor was passed";
-
- int flags = ::fcntl(fd, F_GETFD, 0);
- if (flags < 0) {
-return ErrnoError();
- }
-
- int nflags = cloexec == true ? flags | FD_CLOEXEC
- : flags & ~FD_CLOEXEC;
-
- if (nflags == flags) {
-return Nothing();
- }
-
- if (::fcntl(fd, F_SETFD, nflags) < 0) {
-return ErrnoError();
- }
-
- return Nothing();
-}
-
-
// See `src/libsystemd/sd-daemon/sd-daemon.c` in the systemd source tree
// for the reference implementation. We follow that implementation to
// decide which conditions should result in errors and which should return
@@ -404,7 +380,7 @@ Try> listenFds()
}
for (int fd = SD_LISTEN_FDS_START; fd < SD_LISTEN_FDS_START + n; ++fd) {
-Try cloexec = setCloexecFlag(fd, true);
+Try cloexec = os::cloexec(fd);
if (cloexec.isError()) {
return Error(
"Could not set CLOEXEC flag for file descriptor " + stringify(fd) +
