| // Licensed to the Apache Software Foundation (ASF) under one |
| // or more contributor license agreements. See the NOTICE file |
| // distributed with this work for additional information |
| // regarding copyright ownership. The ASF licenses this file |
| // to you 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 "kudu/server/generic_service.h" |
| |
| #include <ostream> |
| #include <string> |
| #include <unordered_set> |
| #include <utility> |
| // IWYU pragma: no_include <unordered_map> |
| |
| #include <gflags/gflags.h> |
| #include <glog/logging.h> |
| |
| #include "kudu/clock/clock.h" |
| #include "kudu/clock/hybrid_clock.h" |
| #include "kudu/clock/mock_ntp.h" |
| #include "kudu/common/timestamp.h" |
| #include "kudu/common/wire_protocol.h" |
| #include "kudu/gutil/casts.h" |
| #include "kudu/gutil/map-util.h" |
| #include "kudu/gutil/strings/substitute.h" // IWYU pragma: keep |
| #include "kudu/rpc/rpc_context.h" |
| #include "kudu/server/server_base.h" |
| #include "kudu/server/server_base.pb.h" |
| #include "kudu/util/debug-util.h" |
| #include "kudu/util/debug/leak_annotations.h" // IWYU pragma: keep |
| #include "kudu/util/flag_tags.h" |
| #include "kudu/util/flags.h" |
| #include "kudu/util/mem_tracker.h" |
| #include "kudu/util/status.h" |
| |
| DECLARE_string(time_source); |
| DECLARE_bool(use_hybrid_clock); |
| |
| using std::string; |
| using std::unordered_set; |
| using strings::Substitute; |
| |
| namespace kudu { |
| namespace server { |
| |
| GenericServiceImpl::GenericServiceImpl(ServerBase* server) |
| : GenericServiceIf(server->metric_entity(), server->result_tracker()), |
| server_(server) { |
| } |
| |
| GenericServiceImpl::~GenericServiceImpl() { |
| } |
| |
| bool GenericServiceImpl::AuthorizeSuperUser(const google::protobuf::Message* /*req*/, |
| google::protobuf::Message* /*resp*/, |
| rpc::RpcContext* rpc) { |
| return server_->Authorize(rpc, ServerBase::SUPER_USER); |
| } |
| |
| bool GenericServiceImpl::AuthorizeClient(const google::protobuf::Message* /*req*/, |
| google::protobuf::Message* /*resp*/, |
| rpc::RpcContext* rpc) { |
| return server_->Authorize(rpc, ServerBase::SUPER_USER | ServerBase::USER); |
| } |
| |
| |
| void GenericServiceImpl::GetFlags(const GetFlagsRequestPB* req, |
| GetFlagsResponsePB* resp, |
| rpc::RpcContext* rpc) { |
| // If no tags were specified, return all flags that have non-default values. |
| // If flags were specified, will ignore 'all_flags'. |
| // If tags were specified, also filter flags that don't match any tag. |
| bool all_flags = req->all_flags(); |
| unordered_set<string> flags(req->flags().begin(), req->flags().end()); |
| for (const auto& entry : GetFlagsMap()) { |
| if (entry.second.is_default && !all_flags && flags.empty()) { |
| continue; |
| } |
| |
| if (!flags.empty() && !ContainsKey(flags, entry.first)) { |
| continue; |
| } |
| |
| unordered_set<string> tags; |
| GetFlagTags(entry.first, &tags); |
| bool matches = req->tags().empty(); |
| for (const auto& tag : req->tags()) { |
| if (ContainsKey(tags, tag)) { |
| matches = true; |
| break; |
| } |
| } |
| if (!matches) continue; |
| |
| auto* flag = resp->add_flags(); |
| flag->set_name(entry.first); |
| flag->set_value(CheckFlagAndRedact(entry.second, EscapeMode::NONE)); |
| flag->set_is_default_value(entry.second.current_value == entry.second.default_value); |
| for (const auto& tag : tags) { |
| flag->add_tags(tag); |
| } |
| } |
| rpc->RespondSuccess(); |
| } |
| |
| void GenericServiceImpl::SetFlag(const SetFlagRequestPB* req, |
| SetFlagResponsePB* resp, |
| rpc::RpcContext* rpc) { |
| |
| // Validate that the flag exists and get the current value. |
| string old_val; |
| if (!google::GetCommandLineOption(req->flag().c_str(), |
| &old_val)) { |
| resp->set_result(SetFlagResponsePB::NO_SUCH_FLAG); |
| rpc->RespondSuccess(); |
| return; |
| } |
| |
| // Validate that the flag is runtime-changeable. |
| unordered_set<string> tags; |
| GetFlagTags(req->flag(), &tags); |
| if (!ContainsKey(tags, "runtime")) { |
| if (req->force()) { |
| LOG(WARNING) << rpc->requestor_string() << " forcing change of " |
| << "non-runtime-safe flag " << req->flag(); |
| } else { |
| resp->set_result(SetFlagResponsePB::NOT_SAFE); |
| resp->set_msg("Flag is not safe to change at runtime"); |
| rpc->RespondSuccess(); |
| return; |
| } |
| } |
| |
| resp->set_old_value(old_val); |
| |
| // Try to set the new value. |
| string ret = google::SetCommandLineOption( |
| req->flag().c_str(), |
| req->value().c_str()); |
| if (ret.empty()) { |
| resp->set_result(SetFlagResponsePB::BAD_VALUE); |
| resp->set_msg("Unable to set flag: bad value"); |
| rpc->RespondSuccess(); |
| return; |
| } |
| // Validate all flags. |
| if (req->run_consistency_check() && !AreFlagsConsistent()) { |
| string res = google::SetCommandLineOption( |
| req->flag().c_str(), |
| old_val.c_str()); |
| if (ret.empty()) { |
| LOG(WARNING) << Substitute("Failed to roll back flag '$0' to previous value '$1'", |
| req->flag(), old_val); |
| } |
| resp->set_result(SetFlagResponsePB::BAD_VALUE); |
| resp->set_msg("Detected inconsistency in command-line flags"); |
| rpc->RespondSuccess(); |
| return; |
| } |
| LOG(INFO) << rpc->requestor_string() << " changed flags via RPC: " |
| << req->flag() << " from '" << old_val << "' to '" |
| << req->value() << "'"; |
| resp->set_result(SetFlagResponsePB::SUCCESS); |
| resp->set_msg(ret); |
| rpc->RespondSuccess(); |
| } |
| |
| void GenericServiceImpl::CheckLeaks(const CheckLeaksRequestPB* /*req*/, |
| CheckLeaksResponsePB* resp, |
| rpc::RpcContext* rpc) { |
| // We have to use these nested #if statements rather than an && to avoid |
| // a preprocessor error with GCC which doesn't know about __has_feature. |
| #if defined(__has_feature) |
| # if __has_feature(address_sanitizer) |
| # define LSAN_ENABLED |
| # endif |
| #endif |
| #ifndef LSAN_ENABLED |
| resp->set_success(false); |
| #else |
| LOG(INFO) << "Checking for leaks (request via RPC)"; |
| resp->set_success(true); |
| |
| // Workaround for LSAN issue 757 (leak check can give false positives when |
| // run concurrently with other work). If LSAN reports a leak, we'll retry |
| // a few times and see if it goes away on its own. Any real leak would, |
| // by definition, not resolve itself over time. |
| // |
| // See https://github.com/google/sanitizers/issues/757 |
| bool has_leaks = true; |
| for (int i = 0; i < 5; i++) { |
| has_leaks = __lsan_do_recoverable_leak_check(); |
| if (!has_leaks) { |
| if (i > 0) { |
| LOG(WARNING) << strings::Substitute("LeakSanitizer found a leak that went away after $0 " |
| "retries.\n", i) |
| << "Treating it as a false positive and ignoring it."; |
| } |
| break; |
| } |
| SleepFor(MonoDelta::FromMilliseconds(5)); |
| } |
| |
| resp->set_found_leaks(has_leaks); |
| #endif |
| #undef LSAN_ENABLED |
| rpc->RespondSuccess(); |
| } |
| |
| void GenericServiceImpl::FlushCoverage(const FlushCoverageRequestPB* /*req*/, |
| FlushCoverageResponsePB* resp, |
| rpc::RpcContext* rpc) { |
| if (IsCoverageBuild()) { |
| TryFlushCoverage(); |
| LOG(INFO) << "Flushed coverage info. (request from " << rpc->requestor_string() << ")"; |
| resp->set_success(true); |
| } else { |
| LOG(WARNING) << "Non-coverage build cannot flush coverage (request from " |
| << rpc->requestor_string() << ")"; |
| resp->set_success(false); |
| } |
| rpc->RespondSuccess(); |
| } |
| |
| void GenericServiceImpl::ServerClock(const ServerClockRequestPB* req, |
| ServerClockResponsePB* resp, |
| rpc::RpcContext* rpc) { |
| resp->set_timestamp(server_->clock()->Now().ToUint64()); |
| rpc->RespondSuccess(); |
| } |
| |
| void GenericServiceImpl::SetServerWallClockForTests(const SetServerWallClockForTestsRequestPB *req, |
| SetServerWallClockForTestsResponsePB *resp, |
| rpc::RpcContext *context) { |
| if (!FLAGS_use_hybrid_clock || FLAGS_time_source != "mock") { |
| LOG(WARNING) << "Error setting wall clock for tests. Server is not using HybridClock" |
| "or was not started with '--time_source=mock'"; |
| resp->set_success(false); |
| } |
| |
| auto* clock = down_cast<clock::HybridClock*>(server_->clock()); |
| auto* mock = down_cast<clock::MockNtp*>(clock->time_service()); |
| if (req->has_now_usec()) { |
| mock->SetMockClockWallTimeForTests(req->now_usec()); |
| } |
| if (req->has_max_error_usec()) { |
| mock->SetMockMaxClockErrorForTests(req->max_error_usec()); |
| } |
| resp->set_success(true); |
| context->RespondSuccess(); |
| } |
| |
| void GenericServiceImpl::GetStatus(const GetStatusRequestPB* /*req*/, |
| GetStatusResponsePB* resp, |
| rpc::RpcContext* rpc) { |
| // Note: we must ensure that resp->has_status() is true in all cases to |
| // preserve backwards compatibility because it is defined as a required field. |
| Status s = server_->GetStatusPB(resp->mutable_status()); |
| if (!s.ok()) { |
| StatusToPB(s, resp->mutable_error()); |
| } |
| rpc->RespondSuccess(); |
| } |
| |
| void GenericServiceImpl::DumpMemTrackers(const DumpMemTrackersRequestPB* /*req*/, |
| DumpMemTrackersResponsePB* resp, |
| rpc::RpcContext* rpc) { |
| MemTracker::TrackersToPb(resp->mutable_root_tracker()); |
| rpc->RespondSuccess(); |
| } |
| |
| } // namespace server |
| } // namespace kudu |
