Repository: mesos Updated Branches: refs/heads/1.5.x 7dbb70572 -> 539e96942
Sent task (health) check updates over the operator streaming API. With this patch subscribers to the master operator streaming API start receiving task health (check) updates. This allows subscribers to maintain more accurate view of the cluster's state, closer to what the traditional `state.json` endpoint offers. Review: https://reviews.apache.org/r/67320/ (cherry picked from commit 4e6d8047fd127745ef10463818dadc2399732961) Project: http://git-wip-us.apache.org/repos/asf/mesos/repo Commit: http://git-wip-us.apache.org/repos/asf/mesos/commit/ce8af74b Tree: http://git-wip-us.apache.org/repos/asf/mesos/tree/ce8af74b Diff: http://git-wip-us.apache.org/repos/asf/mesos/diff/ce8af74b Branch: refs/heads/1.5.x Commit: ce8af74bfcfdc516f2d4ed592184adfc6f2f96f8 Parents: 7dbb705 Author: Alexander Rukletsov <[email protected]> Authored: Mon Jun 4 16:31:38 2018 +0200 Committer: Alexander Rukletsov <[email protected]> Committed: Mon Jun 4 16:47:13 2018 +0200 ---------------------------------------------------------------------- src/master/master.cpp | 7 +++ src/tests/api_tests.cpp | 130 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 137 insertions(+) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/mesos/blob/ce8af74b/src/master/master.cpp ---------------------------------------------------------------------- diff --git a/src/master/master.cpp b/src/master/master.cpp index d3af45c..6b1764f 100644 --- a/src/master/master.cpp +++ b/src/master/master.cpp @@ -10245,6 +10245,13 @@ void Master::updateTask(Task* task, const StatusUpdate& update) } } + // If this is a (health) check status update, always forward it to + // subscribers. + if (status.reason() == TaskStatus::REASON_TASK_CHECK_STATUS_UPDATED || + status.reason() == TaskStatus::REASON_TASK_HEALTH_CHECK_STATUS_UPDATED) { + sendSubscribersUpdate = true; + } + // TODO(brenden): Consider wiping the `message` field? if (task->statuses_size() > 0 && task->statuses(task->statuses_size() - 1).state() == status.state()) { http://git-wip-us.apache.org/repos/asf/mesos/blob/ce8af74b/src/tests/api_tests.cpp ---------------------------------------------------------------------- diff --git a/src/tests/api_tests.cpp b/src/tests/api_tests.cpp index fb45879..f254532 100644 --- a/src/tests/api_tests.cpp +++ b/src/tests/api_tests.cpp @@ -2268,6 +2268,136 @@ TEST_P(MasterAPITest, Subscribe) } +// This test verifies that subscribers eventually receive events with the +// task health information. +TEST_P(MasterAPITest, SubscribersReceiveHealthUpdates) +{ + ContentType contentType = GetParam(); + + Try<Owned<cluster::Master>> master = StartMaster(); + ASSERT_SOME(master); + + auto scheduler = std::make_shared<v1::MockHTTPScheduler>(); + + Owned<MasterDetector> detector = master.get()->createDetector(); + Try<Owned<cluster::Slave>> slave = StartSlave(detector.get()); + ASSERT_SOME(slave); + + // Connect the scheduler. + Future<Nothing> connected; + EXPECT_CALL(*scheduler, connected(_)) + .WillOnce(FutureSatisfy(&connected)); + + v1::scheduler::TestMesos mesos( + master.get()->pid, + contentType, + scheduler); + + AWAIT_READY(connected); + + // Subscribe the scheduler. This will trigger a resource offer. + Future<v1::scheduler::Event::Subscribed> subscribed; + EXPECT_CALL(*scheduler, subscribed(_, _)) + .WillOnce(FutureArg<1>(&subscribed)); + + EXPECT_CALL(*scheduler, heartbeat(_)) + .WillRepeatedly(Return()); // Ignore heartbeats. + + Future<v1::scheduler::Event::Offers> offers; + EXPECT_CALL(*scheduler, offers(_, _)) + .WillOnce(FutureArg<1>(&offers)); + + mesos.send(v1::createCallSubscribe(v1::DEFAULT_FRAMEWORK_INFO)); + AWAIT_READY(subscribed); + + v1::FrameworkID frameworkId(subscribed->framework_id()); + + // Subscribe to the master's event stream via the v1 API. + v1::master::Call v1Call; + v1Call.set_type(v1::master::Call::SUBSCRIBE); + + http::Headers headers = createBasicAuthHeaders(DEFAULT_CREDENTIAL); + + headers["Accept"] = stringify(contentType); + + Future<http::Response> response = http::streaming::post( + master.get()->pid, + "api/v1", + headers, + serialize(contentType, v1Call), + stringify(contentType)); + + AWAIT_EXPECT_RESPONSE_STATUS_EQ(http::OK().status, response); + AWAIT_EXPECT_RESPONSE_HEADER_EQ("chunked", "Transfer-Encoding", response); + ASSERT_EQ(http::Response::PIPE, response->type); + ASSERT_SOME(response->reader); + + http::Pipe::Reader reader = response->reader.get(); + + auto deserializer = + lambda::bind(deserialize<v1::master::Event>, contentType, lambda::_1); + + Reader<v1::master::Event> decoder( + Decoder<v1::master::Event>(deserializer), reader); + + // Prepare and launch a task with a failing health check using the scheduler. + AWAIT_READY(offers); + ASSERT_FALSE(offers->offers().empty()); + + const v1::Offer& offer = offers->offers(0); + const v1::AgentID agentId(offer.agent_id()); + + TaskInfo task = createTask(devolve(offer), SLEEP_COMMAND(10000)); + + // This describes a single health check that will fail. + HealthCheck healthCheck; + healthCheck.set_type(HealthCheck::HTTP); + healthCheck.mutable_http()->set_port(80); + healthCheck.mutable_http()->set_path("/help"); + healthCheck.set_delay_seconds(0); + healthCheck.set_interval_seconds(1000); + healthCheck.set_grace_period_seconds(0); + task.mutable_health_check()->CopyFrom(healthCheck); + + EXPECT_CALL(*scheduler, update(_, _)) + .WillRepeatedly(v1::scheduler::SendAcknowledge(frameworkId, agentId)); + + mesos.send( + v1::createCallAccept(frameworkId, offer, {v1::LAUNCH({evolve(task)})})); + + // Expect to get a task health update after a _small_ number of other events, + // e.g., `SUBSCRIBED`, initial `HEARTBEAT`, `TASK_STARTING` and `TASK_RUNNING` + // task status updates, etc; 10 seems to be a reasonable number. + // + // TODO(alexr): Instead of guessing the number of events we should filter + // out all uninteresting events. This can be done for example by introducing + // a way to inject a "matcher" for the event, something like: + // AWAIT_READY(event, [](const auto& event ) -> bool { + // return (event->get().type() == ...); }); + int maxEventsToSkip = 10; + bool taskHealthUpdateObserved = false; + while (maxEventsToSkip--) { + Future<Result<v1::master::Event>> event = decoder.read(); + AWAIT_READY(event); + + if (event->get().type() == v1::master::Event::TASK_UPDATED && + event->get().task_updated().status().task_id() == + evolve(task.task_id()) && + event->get().task_updated().status().state() == v1::TASK_RUNNING && + event->get().task_updated().status().reason() == + v1::TaskStatus::REASON_TASK_HEALTH_CHECK_STATUS_UPDATED && + event->get().task_updated().status().healthy() == false) { + taskHealthUpdateObserved = true; + break; + } + } + + ASSERT_TRUE(taskHealthUpdateObserved) + << "Health update for task '" << task.task_id() + << "' has not been received"; +} + + // Verifies that operators subscribed to the master's operator API event // stream only receive events that they are authorized to see. TEST_P(MasterAPITest, EventAuthorizationFiltering)
