mirror of https://github.com/grpc/grpc.git
Merge pull request #15853 from AspirinSJL/load_reporting_service
Add server load reporting servicepull/15851/merge
Juanli Shen
6 years ago
committed by
GitHub
commit
4f840eafec
21 changed files with 1240 additions and 88 deletions
@ -0,0 +1,53 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPCPP_EXT_SERVER_LOAD_REPORTING_H |
||||
#define GRPCPP_EXT_SERVER_LOAD_REPORTING_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <grpc/load_reporting.h> |
||||
#include <grpcpp/impl/codegen/config.h> |
||||
#include <grpcpp/impl/codegen/server_context.h> |
||||
#include <grpcpp/impl/server_builder_option.h> |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
namespace experimental { |
||||
|
||||
// The ServerBuilderOption to enable server-side load reporting feature. To
|
||||
// enable the feature, please make sure the binary builds with the
|
||||
// grpcpp_server_load_reporting library and set this option in the
|
||||
// ServerBuilder.
|
||||
class LoadReportingServiceServerBuilderOption : public ServerBuilderOption { |
||||
public: |
||||
void UpdateArguments(::grpc::ChannelArguments* args) override; |
||||
void UpdatePlugins(std::vector<std::unique_ptr<::grpc::ServerBuilderPlugin>>* |
||||
plugins) override; |
||||
}; |
||||
|
||||
// Adds the load reporting cost with \a cost_name and \a cost_value in the
|
||||
// trailing metadata of the server context.
|
||||
void AddLoadReportingCost(grpc::ServerContext* ctx, |
||||
const grpc::string& cost_name, double cost_value); |
||||
|
||||
} // namespace experimental
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPCPP_EXT_SERVER_LOAD_REPORTING_H
|
||||
@ -0,0 +1,370 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/cpp/server/load_reporter/load_reporter_async_service_impl.h" |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
void LoadReporterAsyncServiceImpl::CallableTag::Run(bool ok) { |
||||
GPR_ASSERT(handler_function_ != nullptr); |
||||
GPR_ASSERT(handler_ != nullptr); |
||||
handler_function_(std::move(handler_), ok); |
||||
} |
||||
|
||||
LoadReporterAsyncServiceImpl::LoadReporterAsyncServiceImpl( |
||||
std::unique_ptr<ServerCompletionQueue> cq) |
||||
: cq_(std::move(cq)) { |
||||
thread_ = std::unique_ptr<::grpc_core::Thread>( |
||||
new ::grpc_core::Thread("server_load_reporting", Work, this)); |
||||
std::unique_ptr<CpuStatsProvider> cpu_stats_provider = nullptr; |
||||
#if defined(GPR_LINUX) || defined(GPR_WINDOWS) || defined(GPR_APPLE) |
||||
cpu_stats_provider.reset(new CpuStatsProviderDefaultImpl()); |
||||
#endif |
||||
load_reporter_ = std::unique_ptr<LoadReporter>(new LoadReporter( |
||||
kFeedbackSampleWindowSeconds, |
||||
std::unique_ptr<CensusViewProvider>(new CensusViewProviderDefaultImpl()), |
||||
std::move(cpu_stats_provider))); |
||||
} |
||||
|
||||
LoadReporterAsyncServiceImpl::~LoadReporterAsyncServiceImpl() { |
||||
// We will reach here after the server starts shutting down.
|
||||
shutdown_ = true; |
||||
{ |
||||
std::unique_lock<std::mutex> lock(cq_shutdown_mu_); |
||||
cq_->Shutdown(); |
||||
} |
||||
if (next_fetch_and_sample_alarm_ != nullptr) |
||||
next_fetch_and_sample_alarm_->Cancel(); |
||||
thread_->Join(); |
||||
} |
||||
|
||||
void LoadReporterAsyncServiceImpl::ScheduleNextFetchAndSample() { |
||||
auto next_fetch_and_sample_time = |
||||
gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), |
||||
gpr_time_from_millis(kFetchAndSampleIntervalSeconds * 1000, |
||||
GPR_TIMESPAN)); |
||||
{ |
||||
std::unique_lock<std::mutex> lock(cq_shutdown_mu_); |
||||
if (shutdown_) return; |
||||
// TODO(juanlishen): Improve the Alarm implementation to reuse a single
|
||||
// instance for multiple events.
|
||||
next_fetch_and_sample_alarm_.reset(new Alarm); |
||||
next_fetch_and_sample_alarm_->Set(cq_.get(), next_fetch_and_sample_time, |
||||
this); |
||||
} |
||||
gpr_log(GPR_DEBUG, "[LRS %p] Next fetch-and-sample scheduled.", this); |
||||
} |
||||
|
||||
void LoadReporterAsyncServiceImpl::FetchAndSample(bool ok) { |
||||
if (!ok) { |
||||
gpr_log(GPR_INFO, "[LRS %p] Fetch-and-sample is stopped.", this); |
||||
return; |
||||
} |
||||
gpr_log(GPR_DEBUG, "[LRS %p] Starting a fetch-and-sample...", this); |
||||
load_reporter_->FetchAndSample(); |
||||
ScheduleNextFetchAndSample(); |
||||
} |
||||
|
||||
void LoadReporterAsyncServiceImpl::Work(void* arg) { |
||||
LoadReporterAsyncServiceImpl* service = |
||||
reinterpret_cast<LoadReporterAsyncServiceImpl*>(arg); |
||||
service->FetchAndSample(true /* ok */); |
||||
// TODO(juanlishen): This is a workaround to wait for the cq to be ready. Need
|
||||
// to figure out why cq is not ready after service starts.
|
||||
gpr_sleep_until(gpr_time_add(gpr_now(GPR_CLOCK_MONOTONIC), |
||||
gpr_time_from_seconds(1, GPR_TIMESPAN))); |
||||
ReportLoadHandler::CreateAndStart(service->cq_.get(), service, |
||||
service->load_reporter_.get()); |
||||
void* tag; |
||||
bool ok; |
||||
while (true) { |
||||
if (!service->cq_->Next(&tag, &ok)) { |
||||
// The completion queue is shutting down.
|
||||
GPR_ASSERT(service->shutdown_); |
||||
break; |
||||
} |
||||
if (tag == service) { |
||||
service->FetchAndSample(ok); |
||||
} else { |
||||
auto* next_step = static_cast<CallableTag*>(tag); |
||||
next_step->Run(ok); |
||||
} |
||||
} |
||||
} |
||||
|
||||
void LoadReporterAsyncServiceImpl::StartThread() { thread_->Start(); } |
||||
|
||||
void LoadReporterAsyncServiceImpl::ReportLoadHandler::CreateAndStart( |
||||
ServerCompletionQueue* cq, LoadReporterAsyncServiceImpl* service, |
||||
LoadReporter* load_reporter) { |
||||
std::shared_ptr<ReportLoadHandler> handler = |
||||
std::make_shared<ReportLoadHandler>(cq, service, load_reporter); |
||||
ReportLoadHandler* p = handler.get(); |
||||
{ |
||||
std::unique_lock<std::mutex> lock(service->cq_shutdown_mu_); |
||||
if (service->shutdown_) return; |
||||
p->on_done_notified_ = |
||||
CallableTag(std::bind(&ReportLoadHandler::OnDoneNotified, p, |
||||
std::placeholders::_1, std::placeholders::_2), |
||||
handler); |
||||
p->next_inbound_ = |
||||
CallableTag(std::bind(&ReportLoadHandler::OnRequestDelivered, p, |
||||
std::placeholders::_1, std::placeholders::_2), |
||||
std::move(handler)); |
||||
p->ctx_.AsyncNotifyWhenDone(&p->on_done_notified_); |
||||
service->RequestReportLoad(&p->ctx_, &p->stream_, cq, cq, |
||||
&p->next_inbound_); |
||||
} |
||||
} |
||||
|
||||
LoadReporterAsyncServiceImpl::ReportLoadHandler::ReportLoadHandler( |
||||
ServerCompletionQueue* cq, LoadReporterAsyncServiceImpl* service, |
||||
LoadReporter* load_reporter) |
||||
: cq_(cq), |
||||
service_(service), |
||||
load_reporter_(load_reporter), |
||||
stream_(&ctx_), |
||||
call_status_(WAITING_FOR_DELIVERY) {} |
||||
|
||||
void LoadReporterAsyncServiceImpl::ReportLoadHandler::OnRequestDelivered( |
||||
std::shared_ptr<ReportLoadHandler> self, bool ok) { |
||||
if (ok) { |
||||
call_status_ = DELIVERED; |
||||
} else { |
||||
// AsyncNotifyWhenDone() needs to be called before the call starts, but the
|
||||
// tag will not pop out if the call never starts (
|
||||
// https://github.com/grpc/grpc/issues/10136). So we need to manually
|
||||
// release the ownership of the handler in this case.
|
||||
GPR_ASSERT(on_done_notified_.ReleaseHandler() != nullptr); |
||||
} |
||||
if (!ok || shutdown_) { |
||||
// The value of ok being false means that the server is shutting down.
|
||||
Shutdown(std::move(self), "OnRequestDelivered"); |
||||
return; |
||||
} |
||||
// Spawn a new handler instance to serve the next new client. Every handler
|
||||
// instance will deallocate itself when it's done.
|
||||
CreateAndStart(cq_, service_, load_reporter_); |
||||
{ |
||||
std::unique_lock<std::mutex> lock(service_->cq_shutdown_mu_); |
||||
if (service_->shutdown_) { |
||||
lock.release()->unlock(); |
||||
Shutdown(std::move(self), "OnRequestDelivered"); |
||||
return; |
||||
} |
||||
next_inbound_ = |
||||
CallableTag(std::bind(&ReportLoadHandler::OnReadDone, this, |
||||
std::placeholders::_1, std::placeholders::_2), |
||||
std::move(self)); |
||||
stream_.Read(&request_, &next_inbound_); |
||||
} |
||||
// LB ID is unique for each load reporting stream.
|
||||
lb_id_ = load_reporter_->GenerateLbId(); |
||||
gpr_log(GPR_INFO, |
||||
"[LRS %p] Call request delivered (lb_id_: %s, handler: %p). " |
||||
"Start reading the initial request...", |
||||
service_, lb_id_.c_str(), this); |
||||
} |
||||
|
||||
void LoadReporterAsyncServiceImpl::ReportLoadHandler::OnReadDone( |
||||
std::shared_ptr<ReportLoadHandler> self, bool ok) { |
||||
if (!ok || shutdown_) { |
||||
if (!ok && call_status_ < INITIAL_REQUEST_RECEIVED) { |
||||
// The client may have half-closed the stream or the stream is broken.
|
||||
gpr_log(GPR_INFO, |
||||
"[LRS %p] Failed reading the initial request from the stream " |
||||
"(lb_id_: %s, handler: %p, done_notified: %d, is_cancelled: %d).", |
||||
service_, lb_id_.c_str(), this, static_cast<int>(done_notified_), |
||||
static_cast<int>(is_cancelled_)); |
||||
} |
||||
Shutdown(std::move(self), "OnReadDone"); |
||||
return; |
||||
} |
||||
// We only receive one request, which is the initial request.
|
||||
if (call_status_ < INITIAL_REQUEST_RECEIVED) { |
||||
if (!request_.has_initial_request()) { |
||||
Shutdown(std::move(self), "OnReadDone+initial_request_not_found"); |
||||
} else { |
||||
call_status_ = INITIAL_REQUEST_RECEIVED; |
||||
const auto& initial_request = request_.initial_request(); |
||||
load_balanced_hostname_ = initial_request.load_balanced_hostname(); |
||||
load_key_ = initial_request.load_key(); |
||||
load_reporter_->ReportStreamCreated(load_balanced_hostname_, lb_id_, |
||||
load_key_); |
||||
const auto& load_report_interval = initial_request.load_report_interval(); |
||||
load_report_interval_ms_ = |
||||
static_cast<uint64_t>(load_report_interval.seconds() * 1000 + |
||||
load_report_interval.nanos() / 1000); |
||||
gpr_log( |
||||
GPR_INFO, |
||||
"[LRS %p] Initial request received. Start load reporting (load " |
||||
"balanced host: %s, interval: %lu ms, lb_id_: %s, handler: %p)...", |
||||
service_, load_balanced_hostname_.c_str(), load_report_interval_ms_, |
||||
lb_id_.c_str(), this); |
||||
SendReport(self, true /* ok */); |
||||
// Expect this read to fail.
|
||||
{ |
||||
std::unique_lock<std::mutex> lock(service_->cq_shutdown_mu_); |
||||
if (service_->shutdown_) { |
||||
lock.release()->unlock(); |
||||
Shutdown(std::move(self), "OnReadDone"); |
||||
return; |
||||
} |
||||
next_inbound_ = |
||||
CallableTag(std::bind(&ReportLoadHandler::OnReadDone, this, |
||||
std::placeholders::_1, std::placeholders::_2), |
||||
std::move(self)); |
||||
stream_.Read(&request_, &next_inbound_); |
||||
} |
||||
} |
||||
} else { |
||||
// Another request received! This violates the spec.
|
||||
gpr_log(GPR_ERROR, |
||||
"[LRS %p] Another request received (lb_id_: %s, handler: %p).", |
||||
service_, lb_id_.c_str(), this); |
||||
Shutdown(std::move(self), "OnReadDone+second_request"); |
||||
} |
||||
} |
||||
|
||||
void LoadReporterAsyncServiceImpl::ReportLoadHandler::ScheduleNextReport( |
||||
std::shared_ptr<ReportLoadHandler> self, bool ok) { |
||||
if (!ok || shutdown_) { |
||||
Shutdown(std::move(self), "ScheduleNextReport"); |
||||
return; |
||||
} |
||||
auto next_report_time = gpr_time_add( |
||||
gpr_now(GPR_CLOCK_MONOTONIC), |
||||
gpr_time_from_millis(load_report_interval_ms_, GPR_TIMESPAN)); |
||||
{ |
||||
std::unique_lock<std::mutex> lock(service_->cq_shutdown_mu_); |
||||
if (service_->shutdown_) { |
||||
lock.release()->unlock(); |
||||
Shutdown(std::move(self), "ScheduleNextReport"); |
||||
return; |
||||
} |
||||
next_outbound_ = |
||||
CallableTag(std::bind(&ReportLoadHandler::SendReport, this, |
||||
std::placeholders::_1, std::placeholders::_2), |
||||
std::move(self)); |
||||
// TODO(juanlishen): Improve the Alarm implementation to reuse a single
|
||||
// instance for multiple events.
|
||||
next_report_alarm_.reset(new Alarm); |
||||
next_report_alarm_->Set(cq_, next_report_time, &next_outbound_); |
||||
} |
||||
gpr_log(GPR_DEBUG, |
||||
"[LRS %p] Next load report scheduled (lb_id_: %s, handler: %p).", |
||||
service_, lb_id_.c_str(), this); |
||||
} |
||||
|
||||
void LoadReporterAsyncServiceImpl::ReportLoadHandler::SendReport( |
||||
std::shared_ptr<ReportLoadHandler> self, bool ok) { |
||||
if (!ok || shutdown_) { |
||||
Shutdown(std::move(self), "SendReport"); |
||||
return; |
||||
} |
||||
::grpc::lb::v1::LoadReportResponse response; |
||||
auto loads = load_reporter_->GenerateLoads(load_balanced_hostname_, lb_id_); |
||||
response.mutable_load()->Swap(&loads); |
||||
auto feedback = load_reporter_->GenerateLoadBalancingFeedback(); |
||||
response.mutable_load_balancing_feedback()->Swap(&feedback); |
||||
if (call_status_ < INITIAL_RESPONSE_SENT) { |
||||
auto initial_response = response.mutable_initial_response(); |
||||
initial_response->set_load_balancer_id(lb_id_); |
||||
initial_response->set_implementation_id( |
||||
::grpc::lb::v1::InitialLoadReportResponse::CPP); |
||||
initial_response->set_server_version(kVersion); |
||||
call_status_ = INITIAL_RESPONSE_SENT; |
||||
} |
||||
{ |
||||
std::unique_lock<std::mutex> lock(service_->cq_shutdown_mu_); |
||||
if (service_->shutdown_) { |
||||
lock.release()->unlock(); |
||||
Shutdown(std::move(self), "SendReport"); |
||||
return; |
||||
} |
||||
next_outbound_ = |
||||
CallableTag(std::bind(&ReportLoadHandler::ScheduleNextReport, this, |
||||
std::placeholders::_1, std::placeholders::_2), |
||||
std::move(self)); |
||||
stream_.Write(response, &next_outbound_); |
||||
gpr_log(GPR_INFO, |
||||
"[LRS %p] Sending load report (lb_id_: %s, handler: %p, loads " |
||||
"count: %d)...", |
||||
service_, lb_id_.c_str(), this, response.load().size()); |
||||
} |
||||
} |
||||
|
||||
void LoadReporterAsyncServiceImpl::ReportLoadHandler::OnDoneNotified( |
||||
std::shared_ptr<ReportLoadHandler> self, bool ok) { |
||||
GPR_ASSERT(ok); |
||||
done_notified_ = true; |
||||
if (ctx_.IsCancelled()) { |
||||
is_cancelled_ = true; |
||||
} |
||||
gpr_log(GPR_INFO, |
||||
"[LRS %p] Load reporting call is notified done (handler: %p, " |
||||
"is_cancelled: %d).", |
||||
service_, this, static_cast<int>(is_cancelled_)); |
||||
Shutdown(std::move(self), "OnDoneNotified"); |
||||
} |
||||
|
||||
void LoadReporterAsyncServiceImpl::ReportLoadHandler::Shutdown( |
||||
std::shared_ptr<ReportLoadHandler> self, const char* reason) { |
||||
if (!shutdown_) { |
||||
gpr_log(GPR_INFO, |
||||
"[LRS %p] Shutting down the handler (lb_id_: %s, handler: %p, " |
||||
"reason: %s).", |
||||
service_, lb_id_.c_str(), this, reason); |
||||
shutdown_ = true; |
||||
if (call_status_ >= INITIAL_REQUEST_RECEIVED) { |
||||
load_reporter_->ReportStreamClosed(load_balanced_hostname_, lb_id_); |
||||
next_report_alarm_->Cancel(); |
||||
} |
||||
} |
||||
// OnRequestDelivered() may be called after OnDoneNotified(), so we need to
|
||||
// try to Finish() every time we are in Shutdown().
|
||||
if (call_status_ >= DELIVERED && call_status_ < FINISH_CALLED) { |
||||
std::unique_lock<std::mutex> lock(service_->cq_shutdown_mu_); |
||||
if (!service_->shutdown_) { |
||||
on_finish_done_ = |
||||
CallableTag(std::bind(&ReportLoadHandler::OnFinishDone, this, |
||||
std::placeholders::_1, std::placeholders::_2), |
||||
std::move(self)); |
||||
// TODO(juanlishen): Maybe add a message proto for the client to
|
||||
// explicitly cancel the stream so that we can return OK status in such
|
||||
// cases.
|
||||
stream_.Finish(Status::CANCELLED, &on_finish_done_); |
||||
call_status_ = FINISH_CALLED; |
||||
} |
||||
} |
||||
} |
||||
|
||||
void LoadReporterAsyncServiceImpl::ReportLoadHandler::OnFinishDone( |
||||
std::shared_ptr<ReportLoadHandler> self, bool ok) { |
||||
if (ok) { |
||||
gpr_log(GPR_INFO, |
||||
"[LRS %p] Load reporting finished (lb_id_: %s, handler: %p).", |
||||
service_, lb_id_.c_str(), this); |
||||
} |
||||
} |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,194 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_SRC_CPP_SERVER_LOAD_REPORTER_ASYNC_SERVICE_IMPL_H |
||||
#define GRPC_SRC_CPP_SERVER_LOAD_REPORTER_ASYNC_SERVICE_IMPL_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <grpc/support/log.h> |
||||
#include <grpcpp/alarm.h> |
||||
#include <grpcpp/grpcpp.h> |
||||
|
||||
#include "src/core/lib/gprpp/thd.h" |
||||
#include "src/cpp/server/load_reporter/load_reporter.h" |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
// Async load reporting service. It's mainly responsible for controlling the
|
||||
// procedure of incoming requests. The real business logic is handed off to the
|
||||
// LoadReporter. There should be at most one instance of this service on a
|
||||
// server to avoid spreading the load data into multiple places.
|
||||
class LoadReporterAsyncServiceImpl |
||||
: public grpc::lb::v1::LoadReporter::AsyncService { |
||||
public: |
||||
explicit LoadReporterAsyncServiceImpl( |
||||
std::unique_ptr<ServerCompletionQueue> cq); |
||||
~LoadReporterAsyncServiceImpl(); |
||||
|
||||
// Starts the working thread.
|
||||
void StartThread(); |
||||
|
||||
// Not copyable nor movable.
|
||||
LoadReporterAsyncServiceImpl(const LoadReporterAsyncServiceImpl&) = delete; |
||||
LoadReporterAsyncServiceImpl& operator=(const LoadReporterAsyncServiceImpl&) = |
||||
delete; |
||||
|
||||
private: |
||||
class ReportLoadHandler; |
||||
|
||||
// A tag that can be called with a bool argument. It's tailored for
|
||||
// ReportLoadHandler's use. Before being used, it should be constructed with a
|
||||
// method of ReportLoadHandler and a shared pointer to the handler. The
|
||||
// shared pointer will be moved to the invoked function and the function can
|
||||
// only be invoked once. That makes ref counting of the handler easier,
|
||||
// because the shared pointer is not bound to the function and can be gone
|
||||
// once the invoked function returns (if not used any more).
|
||||
class CallableTag { |
||||
public: |
||||
using HandlerFunction = |
||||
std::function<void(std::shared_ptr<ReportLoadHandler>, bool)>; |
||||
|
||||
CallableTag() {} |
||||
|
||||
CallableTag(HandlerFunction func, |
||||
std::shared_ptr<ReportLoadHandler> handler) |
||||
: handler_function_(std::move(func)), handler_(std::move(handler)) { |
||||
GPR_ASSERT(handler_function_ != nullptr); |
||||
GPR_ASSERT(handler_ != nullptr); |
||||
} |
||||
|
||||
// Runs the tag. This should be called only once. The handler is no longer
|
||||
// owned by this tag after this method is invoked.
|
||||
void Run(bool ok); |
||||
|
||||
// Releases and returns the shared pointer to the handler.
|
||||
std::shared_ptr<ReportLoadHandler> ReleaseHandler() { |
||||
return std::move(handler_); |
||||
} |
||||
|
||||
private: |
||||
HandlerFunction handler_function_ = nullptr; |
||||
std::shared_ptr<ReportLoadHandler> handler_; |
||||
}; |
||||
|
||||
// Each handler takes care of one load reporting stream. It contains
|
||||
// per-stream data and it will access the members of the parent class (i.e.,
|
||||
// LoadReporterAsyncServiceImpl) for service-wide data (e.g., the load data).
|
||||
class ReportLoadHandler { |
||||
public: |
||||
// Instantiates a ReportLoadHandler and requests the next load reporting
|
||||
// call. The handler object will manage its own lifetime, so no action is
|
||||
// needed from the caller any more regarding that object.
|
||||
static void CreateAndStart(ServerCompletionQueue* cq, |
||||
LoadReporterAsyncServiceImpl* service, |
||||
LoadReporter* load_reporter); |
||||
|
||||
// This ctor is public because we want to use std::make_shared<> in
|
||||
// CreateAndStart(). This ctor shouldn't be used elsewhere.
|
||||
ReportLoadHandler(ServerCompletionQueue* cq, |
||||
LoadReporterAsyncServiceImpl* service, |
||||
LoadReporter* load_reporter); |
||||
|
||||
private: |
||||
// After the handler has a call request delivered, it starts reading the
|
||||
// initial request. Also, a new handler is spawned so that we can keep
|
||||
// servicing future calls.
|
||||
void OnRequestDelivered(std::shared_ptr<ReportLoadHandler> self, bool ok); |
||||
|
||||
// The first Read() is expected to succeed, after which the handler starts
|
||||
// sending load reports back to the balancer. The second Read() is
|
||||
// expected to fail, which happens when the balancer half-closes the
|
||||
// stream to signal that it's no longer interested in the load reports. For
|
||||
// the latter case, the handler will then close the stream.
|
||||
void OnReadDone(std::shared_ptr<ReportLoadHandler> self, bool ok); |
||||
|
||||
// The report sending operations are sequential as: send report -> send
|
||||
// done, schedule the next send -> waiting for the alarm to fire -> alarm
|
||||
// fires, send report -> ...
|
||||
void SendReport(std::shared_ptr<ReportLoadHandler> self, bool ok); |
||||
void ScheduleNextReport(std::shared_ptr<ReportLoadHandler> self, bool ok); |
||||
|
||||
// Called when Finish() is done.
|
||||
void OnFinishDone(std::shared_ptr<ReportLoadHandler> self, bool ok); |
||||
|
||||
// Called when AsyncNotifyWhenDone() notifies us.
|
||||
void OnDoneNotified(std::shared_ptr<ReportLoadHandler> self, bool ok); |
||||
|
||||
void Shutdown(std::shared_ptr<ReportLoadHandler> self, const char* reason); |
||||
|
||||
// The key fields of the stream.
|
||||
grpc::string lb_id_; |
||||
grpc::string load_balanced_hostname_; |
||||
grpc::string load_key_; |
||||
uint64_t load_report_interval_ms_; |
||||
|
||||
// The data for RPC communication with the load reportee.
|
||||
ServerContext ctx_; |
||||
::grpc::lb::v1::LoadReportRequest request_; |
||||
|
||||
// The members passed down from LoadReporterAsyncServiceImpl.
|
||||
ServerCompletionQueue* cq_; |
||||
LoadReporterAsyncServiceImpl* service_; |
||||
LoadReporter* load_reporter_; |
||||
ServerAsyncReaderWriter<::grpc::lb::v1::LoadReportResponse, |
||||
::grpc::lb::v1::LoadReportRequest> |
||||
stream_; |
||||
|
||||
// The status of the RPC progress.
|
||||
enum CallStatus { |
||||
WAITING_FOR_DELIVERY, |
||||
DELIVERED, |
||||
INITIAL_REQUEST_RECEIVED, |
||||
INITIAL_RESPONSE_SENT, |
||||
FINISH_CALLED |
||||
} call_status_; |
||||
bool shutdown_{false}; |
||||
bool done_notified_{false}; |
||||
bool is_cancelled_{false}; |
||||
CallableTag on_done_notified_; |
||||
CallableTag on_finish_done_; |
||||
CallableTag next_inbound_; |
||||
CallableTag next_outbound_; |
||||
std::unique_ptr<Alarm> next_report_alarm_; |
||||
}; |
||||
|
||||
// Handles the incoming requests and drives the completion queue in a loop.
|
||||
static void Work(void* arg); |
||||
|
||||
// Schedules the next data fetching from Census and LB feedback sampling.
|
||||
void ScheduleNextFetchAndSample(); |
||||
|
||||
// Fetches data from Census and samples LB feedback.
|
||||
void FetchAndSample(bool ok); |
||||
|
||||
std::unique_ptr<ServerCompletionQueue> cq_; |
||||
// To synchronize the operations related to shutdown state of cq_, so that we
|
||||
// don't enqueue new tags into cq_ after it is already shut down.
|
||||
std::mutex cq_shutdown_mu_; |
||||
std::atomic_bool shutdown_{false}; |
||||
std::unique_ptr<::grpc_core::Thread> thread_; |
||||
std::unique_ptr<LoadReporter> load_reporter_; |
||||
std::unique_ptr<Alarm> next_fetch_and_sample_alarm_; |
||||
}; |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPC_SRC_CPP_SERVER_LOAD_REPORTER_ASYNC_SERVICE_IMPL_H
|
||||
@ -0,0 +1,41 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <grpcpp/ext/server_load_reporting.h> |
||||
|
||||
#include "src/cpp/server/load_reporter/load_reporting_service_server_builder_plugin.h" |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
namespace experimental { |
||||
|
||||
void LoadReportingServiceServerBuilderOption::UpdateArguments( |
||||
::grpc::ChannelArguments* args) { |
||||
args->SetInt(GRPC_ARG_ENABLE_LOAD_REPORTING, true); |
||||
} |
||||
|
||||
void LoadReportingServiceServerBuilderOption::UpdatePlugins( |
||||
std::vector<std::unique_ptr<::grpc::ServerBuilderPlugin>>* plugins) { |
||||
plugins->emplace_back(new LoadReportingServiceServerBuilderPlugin()); |
||||
} |
||||
|
||||
} // namespace experimental
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,60 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include "src/cpp/server/load_reporter/load_reporting_service_server_builder_plugin.h" |
||||
|
||||
#include <grpcpp/impl/server_initializer.h> |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
bool LoadReportingServiceServerBuilderPlugin::has_sync_methods() const { |
||||
if (service_ != nullptr) { |
||||
return service_->has_synchronous_methods(); |
||||
} |
||||
return false; |
||||
} |
||||
|
||||
bool LoadReportingServiceServerBuilderPlugin::has_async_methods() const { |
||||
if (service_ != nullptr) { |
||||
return service_->has_async_methods(); |
||||
} |
||||
return false; |
||||
} |
||||
|
||||
void LoadReportingServiceServerBuilderPlugin::UpdateServerBuilder( |
||||
grpc::ServerBuilder* builder) { |
||||
auto cq = builder->AddCompletionQueue(); |
||||
service_ = std::make_shared<LoadReporterAsyncServiceImpl>(std::move(cq)); |
||||
} |
||||
|
||||
void LoadReportingServiceServerBuilderPlugin::InitServer( |
||||
grpc::ServerInitializer* si) { |
||||
si->RegisterService(service_); |
||||
} |
||||
|
||||
void LoadReportingServiceServerBuilderPlugin::Finish( |
||||
grpc::ServerInitializer* si) { |
||||
service_->StartThread(); |
||||
service_.reset(); |
||||
} |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,62 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#ifndef GRPC_SRC_CPP_LOAD_REPORTING_SERVICE_SERVER_BUILDER_PLUGIN_H |
||||
#define GRPC_SRC_CPP_LOAD_REPORTING_SERVICE_SERVER_BUILDER_PLUGIN_H |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <grpcpp/impl/server_builder_plugin.h> |
||||
|
||||
#include "src/cpp/server/load_reporter/load_reporter_async_service_impl.h" |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
|
||||
// The plugin that registers and starts load reporting service when starting a
|
||||
// server.
|
||||
class LoadReportingServiceServerBuilderPlugin : public ServerBuilderPlugin { |
||||
public: |
||||
~LoadReportingServiceServerBuilderPlugin() override = default; |
||||
grpc::string name() override { return "load_reporting_service"; } |
||||
|
||||
// Creates a load reporting service.
|
||||
void UpdateServerBuilder(grpc::ServerBuilder* builder) override; |
||||
|
||||
// Registers the load reporter service.
|
||||
void InitServer(grpc::ServerInitializer* si) override; |
||||
|
||||
// Starts the load reporter service.
|
||||
void Finish(grpc::ServerInitializer* si) override; |
||||
|
||||
void ChangeArguments(const grpc::string& name, void* value) override {} |
||||
void UpdateChannelArguments(grpc::ChannelArguments* args) override {} |
||||
bool has_sync_methods() const override; |
||||
bool has_async_methods() const override; |
||||
|
||||
private: |
||||
std::shared_ptr<LoadReporterAsyncServiceImpl> service_; |
||||
}; |
||||
|
||||
std::unique_ptr<grpc::ServerBuilderPlugin> |
||||
CreateLoadReportingServiceServerBuilderPlugin(); |
||||
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
|
||||
#endif // GRPC_SRC_CPP_LOAD_REPORTING_SERVICE_SERVER_BUILDER_PLUGIN_H
|
||||
@ -0,0 +1,45 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/impl/codegen/port_platform.h> |
||||
|
||||
#include <grpcpp/ext/server_load_reporting.h> |
||||
|
||||
#include <grpc/support/log.h> |
||||
|
||||
namespace grpc { |
||||
namespace load_reporter { |
||||
namespace experimental { |
||||
|
||||
void AddLoadReportingCost(grpc::ServerContext* ctx, |
||||
const grpc::string& cost_name, double cost_value) { |
||||
if (std::isnormal(cost_value)) { |
||||
grpc::string buf; |
||||
buf.resize(sizeof(cost_value) + cost_name.size()); |
||||
memcpy(&(*buf.begin()), &cost_value, sizeof(cost_value)); |
||||
memcpy(&(*buf.begin()) + sizeof(cost_value), cost_name.data(), |
||||
cost_name.size()); |
||||
ctx->AddTrailingMetadata(GRPC_LB_COST_MD_KEY, buf); |
||||
} else { |
||||
gpr_log(GPR_ERROR, "Call metric value is not normal."); |
||||
} |
||||
} |
||||
|
||||
} // namespace experimental
|
||||
} // namespace load_reporter
|
||||
} // namespace grpc
|
||||
@ -0,0 +1,191 @@ |
||||
/*
|
||||
* |
||||
* Copyright 2018 gRPC authors. |
||||
* |
||||
* Licensed under the Apache License, Version 2.0 (the "License"); |
||||
* you may not use this file except in compliance with the License. |
||||
* You may obtain a copy of the License at |
||||
* |
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
* |
||||
* Unless required by applicable law or agreed to in writing, software |
||||
* distributed under the License is distributed on an "AS IS" BASIS, |
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
||||
* See the License for the specific language governing permissions and |
||||
* limitations under the License. |
||||
* |
||||
*/ |
||||
|
||||
#include <grpc/support/port_platform.h> |
||||
|
||||
#include <thread> |
||||
|
||||
#include <gmock/gmock.h> |
||||
#include <gtest/gtest.h> |
||||
|
||||
#include <grpc++/grpc++.h> |
||||
#include <grpc/grpc.h> |
||||
#include <grpc/support/log.h> |
||||
#include <grpc/support/string_util.h> |
||||
#include <grpcpp/ext/server_load_reporting.h> |
||||
#include <grpcpp/server_builder.h> |
||||
|
||||
#include "src/proto/grpc/lb/v1/load_reporter.grpc.pb.h" |
||||
#include "src/proto/grpc/testing/echo.grpc.pb.h" |
||||
#include "test/core/util/port.h" |
||||
|
||||
namespace grpc { |
||||
namespace testing { |
||||
namespace { |
||||
|
||||
constexpr double kMetricValue = 3.1415; |
||||
constexpr char kMetricName[] = "METRIC_PI"; |
||||
|
||||
// Different messages result in different response statuses. For simplicity in
|
||||
// computing request bytes, the message sizes should be the same.
|
||||
const char kOkMessage[] = "hello"; |
||||
const char kServerErrorMessage[] = "sverr"; |
||||
const char kClientErrorMessage[] = "clerr"; |
||||
|
||||
class EchoTestServiceImpl : public EchoTestService::Service { |
||||
public: |
||||
~EchoTestServiceImpl() override {} |
||||
|
||||
Status Echo(ServerContext* context, const EchoRequest* request, |
||||
EchoResponse* response) override { |
||||
if (request->message() == kServerErrorMessage) { |
||||
return Status(StatusCode::UNKNOWN, "Server error requested"); |
||||
} |
||||
if (request->message() == kClientErrorMessage) { |
||||
return Status(StatusCode::FAILED_PRECONDITION, "Client error requested"); |
||||
} |
||||
response->set_message(request->message()); |
||||
::grpc::load_reporter::experimental::AddLoadReportingCost( |
||||
context, kMetricName, kMetricValue); |
||||
return Status::OK; |
||||
} |
||||
}; |
||||
|
||||
class ServerLoadReportingEnd2endTest : public ::testing::Test { |
||||
protected: |
||||
void SetUp() override { |
||||
server_address_ = |
||||
"localhost:" + std::to_string(grpc_pick_unused_port_or_die()); |
||||
server_ = |
||||
ServerBuilder() |
||||
.AddListeningPort(server_address_, InsecureServerCredentials()) |
||||
.RegisterService(&echo_service_) |
||||
.SetOption(std::unique_ptr<::grpc::ServerBuilderOption>( |
||||
new ::grpc::load_reporter::experimental:: |
||||
LoadReportingServiceServerBuilderOption())) |
||||
.BuildAndStart(); |
||||
server_thread_ = |
||||
std::thread(&ServerLoadReportingEnd2endTest::RunServerLoop, this); |
||||
} |
||||
|
||||
void RunServerLoop() { server_->Wait(); } |
||||
|
||||
void TearDown() override { |
||||
server_->Shutdown(); |
||||
server_thread_.join(); |
||||
} |
||||
|
||||
void ClientMakeEchoCalls(const grpc::string& lb_id, |
||||
const grpc::string& lb_tag, |
||||
const grpc::string& message, size_t num_requests) { |
||||
auto stub = EchoTestService::NewStub( |
||||
CreateChannel(server_address_, InsecureChannelCredentials())); |
||||
grpc::string lb_token = lb_id + lb_tag; |
||||
for (int i = 0; i < num_requests; ++i) { |
||||
ClientContext ctx; |
||||
if (!lb_token.empty()) ctx.AddMetadata(GRPC_LB_TOKEN_MD_KEY, lb_token); |
||||
EchoRequest request; |
||||
EchoResponse response; |
||||
request.set_message(message); |
||||
Status status = stub->Echo(&ctx, request, &response); |
||||
if (message == kOkMessage) { |
||||
ASSERT_EQ(status.error_code(), StatusCode::OK); |
||||
ASSERT_EQ(request.message(), response.message()); |
||||
} else if (message == kServerErrorMessage) { |
||||
ASSERT_EQ(status.error_code(), StatusCode::UNKNOWN); |
||||
} else if (message == kClientErrorMessage) { |
||||
ASSERT_EQ(status.error_code(), StatusCode::FAILED_PRECONDITION); |
||||
} |
||||
} |
||||
} |
||||
|
||||
grpc::string server_address_; |
||||
std::unique_ptr<Server> server_; |
||||
std::thread server_thread_; |
||||
EchoTestServiceImpl echo_service_; |
||||
}; |
||||
|
||||
TEST_F(ServerLoadReportingEnd2endTest, NoCall) {} |
||||
|
||||
TEST_F(ServerLoadReportingEnd2endTest, BasicReport) { |
||||
auto channel = |
||||
grpc::CreateChannel(server_address_, InsecureChannelCredentials()); |
||||
auto stub = ::grpc::lb::v1::LoadReporter::NewStub(channel); |
||||
ClientContext ctx; |
||||
auto stream = stub->ReportLoad(&ctx); |
||||
::grpc::lb::v1::LoadReportRequest request; |
||||
request.mutable_initial_request()->set_load_balanced_hostname( |
||||
server_address_); |
||||
request.mutable_initial_request()->set_load_key("LOAD_KEY"); |
||||
request.mutable_initial_request() |
||||
->mutable_load_report_interval() |
||||
->set_seconds(5); |
||||
stream->Write(request); |
||||
gpr_log(GPR_INFO, "Initial request sent."); |
||||
::grpc::lb::v1::LoadReportResponse response; |
||||
stream->Read(&response); |
||||
const grpc::string& lb_id = response.initial_response().load_balancer_id(); |
||||
gpr_log(GPR_INFO, "Initial response received (lb_id: %s).", lb_id.c_str()); |
||||
ClientMakeEchoCalls(lb_id, "LB_TAG", kOkMessage, 1); |
||||
while (true) { |
||||
stream->Read(&response); |
||||
if (!response.load().empty()) { |
||||
ASSERT_EQ(response.load().size(), 3); |
||||
for (const auto& load : response.load()) { |
||||
if (load.in_progress_report_case()) { |
||||
// The special load record that reports the number of in-progress
|
||||
// calls.
|
||||
ASSERT_EQ(load.num_calls_in_progress(), 1); |
||||
} else if (load.orphaned_load_case()) { |
||||
// The call from the balancer doesn't have any valid LB token.
|
||||
ASSERT_EQ(load.orphaned_load_case(), load.kLoadKeyUnknown); |
||||
ASSERT_EQ(load.num_calls_started(), 1); |
||||
ASSERT_EQ(load.num_calls_finished_without_error(), 0); |
||||
ASSERT_EQ(load.num_calls_finished_with_error(), 0); |
||||
} else { |
||||
// This corresponds to the calls from the client.
|
||||
ASSERT_EQ(load.num_calls_started(), 1); |
||||
ASSERT_EQ(load.num_calls_finished_without_error(), 1); |
||||
ASSERT_EQ(load.num_calls_finished_with_error(), 0); |
||||
ASSERT_GE(load.total_bytes_received(), sizeof(kOkMessage)); |
||||
ASSERT_GE(load.total_bytes_sent(), sizeof(kOkMessage)); |
||||
ASSERT_EQ(load.metric_data().size(), 1); |
||||
ASSERT_EQ(load.metric_data().Get(0).metric_name(), kMetricName); |
||||
ASSERT_EQ(load.metric_data().Get(0).num_calls_finished_with_metric(), |
||||
1); |
||||
ASSERT_EQ(load.metric_data().Get(0).total_metric_value(), |
||||
kMetricValue); |
||||
} |
||||
} |
||||
break; |
||||
} |
||||
} |
||||
stream->WritesDone(); |
||||
ASSERT_EQ(stream->Finish().error_code(), StatusCode::CANCELLED); |
||||
} |
||||
|
||||
// TODO(juanlishen): Add more tests.
|
||||
|
||||
} // namespace
|
||||
} // namespace testing
|
||||
} // namespace grpc
|
||||
|
||||
int main(int argc, char** argv) { |
||||
::testing::InitGoogleTest(&argc, argv); |
||||
return RUN_ALL_TESTS(); |
||||
} |
||||
Loading…
Reference in new issue