You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by to...@apache.org on 2016/03/15 17:19:02 UTC
[2/2] incubator-kudu git commit: rpc: call-level feature flags
rpc: call-level feature flags
This adds a new API RpcController::RequireServerFeature on the client side, and
ServiceIf::SupportsFeature on the service side. These APIs make it possible for
the client to require application-specific feature support on the server in
order for an individual RPC call to be serviced. If the feature is not
supported, the RPC is rejected and an appropriate error is returned.
This will be used in the future when adding features to existing RPC messages,
so that the server does not silently ignore additional fields.
These RPC features require a new RPC feature flag, APPLICATION_FEATURE_FLAGS, to
be supported. If the server does not support that RPC feature flag, then the
call will fail on the client side.
Change-Id: I46e46bcf80b93fc6cce50f37dd71a6e6539047f8
Reviewed-on: http://gerrit.cloudera.org:8080/2239
Tested-by: Kudu Jenkins
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-kudu/commit/f82919ed
Tree: http://git-wip-us.apache.org/repos/asf/incubator-kudu/tree/f82919ed
Diff: http://git-wip-us.apache.org/repos/asf/incubator-kudu/diff/f82919ed
Branch: refs/heads/master
Commit: f82919ed595a415053b7f4180f70a59674a32730
Parents: 86a510a
Author: Todd Lipcon <to...@apache.org>
Authored: Thu Feb 18 17:55:18 2016 -0800
Committer: Todd Lipcon <to...@apache.org>
Committed: Tue Mar 15 16:15:50 2016 +0000
----------------------------------------------------------------------
src/kudu/rpc/connection.cc | 36 ++++++++++++---
src/kudu/rpc/constants.cc | 4 +-
src/kudu/rpc/inbound_call.cc | 23 ++++++++-
src/kudu/rpc/inbound_call.h | 6 +++
src/kudu/rpc/outbound_call.cc | 24 ++++++++--
src/kudu/rpc/outbound_call.h | 3 ++
src/kudu/rpc/rpc-test-base.h | 42 ++++++++---------
src/kudu/rpc/rpc-test.cc | 77 +++++++++++++++++++++++++++++++
src/kudu/rpc/rpc_controller.cc | 5 ++
src/kudu/rpc/rpc_controller.h | 51 ++++++++++++++++++++
src/kudu/rpc/rpc_header.proto | 25 ++++++++--
src/kudu/rpc/rtest.proto | 5 ++
src/kudu/rpc/sasl_rpc-test.cc | 4 +-
src/kudu/rpc/service_if.cc | 4 ++
src/kudu/rpc/service_if.h | 4 ++
src/kudu/rpc/service_pool.cc | 12 +++++
src/kudu/rpc/transfer.cc | 7 ++-
src/kudu/rpc/transfer.h | 26 +++++++++--
src/kudu/util/CMakeLists.txt | 3 +-
src/kudu/util/scoped_cleanup-test.cc | 34 ++++++++++++++
src/kudu/util/scoped_cleanup.h | 40 ++++++++++++++++
21 files changed, 384 insertions(+), 51 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/connection.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/connection.cc b/src/kudu/rpc/connection.cc
index f1626ea..b811911 100644
--- a/src/kudu/rpc/connection.cc
+++ b/src/kudu/rpc/connection.cc
@@ -17,12 +17,13 @@
#include "kudu/rpc/connection.h"
+#include <algorithm>
#include <boost/intrusive/list.hpp>
#include <gflags/gflags.h>
#include <glog/logging.h>
-#include <stdint.h>
-
#include <iostream>
+#include <set>
+#include <stdint.h>
#include <string>
#include <vector>
@@ -44,6 +45,9 @@
#include "kudu/util/status.h"
#include "kudu/util/trace.h"
+using std::function;
+using std::includes;
+using std::set;
using std::shared_ptr;
using std::vector;
using strings::Substitute;
@@ -214,7 +218,6 @@ void Connection::HandleOutboundCallTimeout(CallAwaitingResponse *car) {
// already timed out.
}
-
// Callbacks after sending a call on the wire.
// This notifies the OutboundCall object to change its state to SENT once it
// has been fully transmitted.
@@ -291,7 +294,7 @@ void Connection::QueueOutboundCall(const shared_ptr<OutboundCall> &call) {
TransferCallbacks *cb = new CallTransferCallbacks(call);
awaiting_response_[call_id] = car.release();
QueueOutbound(gscoped_ptr<OutboundTransfer>(
- new OutboundTransfer(slices_tmp_, cb)));
+ new OutboundTransfer(call_id, slices_tmp_, call->RequiredRpcFeatures(), cb)));
}
// Callbacks for sending an RPC call response from the server.
@@ -367,7 +370,9 @@ void Connection::QueueResponseForCall(gscoped_ptr<InboundCall> call) {
TransferCallbacks *cb = new ResponseTransferCallbacks(std::move(call), this);
// After the response is sent, can delete the InboundCall object.
- gscoped_ptr<OutboundTransfer> t(new OutboundTransfer(slices, cb));
+ // We set a dummy call ID and required feature set, since these are not needed
+ // when sending responses.
+ gscoped_ptr<OutboundTransfer> t(new OutboundTransfer(-1, slices, {}, cb));
QueueTransferTask *task = new QueueTransferTask(std::move(t), this);
reactor_thread_->reactor()->ScheduleReactorTask(task);
@@ -497,6 +502,26 @@ void Connection::WriteHandler(ev::io &watcher, int revents) {
while (!outbound_transfers_.empty()) {
transfer = &(outbound_transfers_.front());
+ if (!transfer->TransferStarted()) {
+ // If this is the start of the transfer, then check if the server has the
+ // required RPC flags. We have to wait until just before the transfer in
+ // order to ensure that the negotiation has taken place, so that the flags
+ // are available.
+ const set<RpcFeatureFlag>& required_features = transfer->required_features();
+ const set<RpcFeatureFlag>& server_features = sasl_client_.server_features();
+ if (!includes(server_features.begin(), server_features.end(),
+ required_features.begin(), required_features.end())) {
+ outbound_transfers_.pop_front();
+ CallAwaitingResponse* car = FindOrDie(awaiting_response_, transfer->call_id());
+ Status s = Status::NotSupported("server does not support the required RPC features");
+ transfer->Abort(s);
+ car->call->SetFailed(s);
+ car->call.reset();
+ delete transfer;
+ continue;
+ }
+ }
+
last_activity_time_ = reactor_thread_->cur_time();
Status status = transfer->SendBuffer(socket_);
if (PREDICT_FALSE(!status.ok())) {
@@ -514,7 +539,6 @@ void Connection::WriteHandler(ev::io &watcher, int revents) {
delete transfer;
}
-
// If we were able to write all of our outbound transfers,
// we don't have any more to write.
write_io_.stop();
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/constants.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/constants.cc b/src/kudu/rpc/constants.cc
index 68955e8..c86bc29 100644
--- a/src/kudu/rpc/constants.cc
+++ b/src/kudu/rpc/constants.cc
@@ -25,8 +25,8 @@ namespace rpc {
const char* const kMagicNumber = "hrpc";
const char* const kSaslAppName = "Kudu";
const char* const kSaslProtoName = "kudu";
-set<RpcFeatureFlag> kSupportedServerRpcFeatureFlags = { TMP_TEST_FEATURE_FLAG };
-set<RpcFeatureFlag> kSupportedClientRpcFeatureFlags = { TMP_TEST_FEATURE_FLAG };
+set<RpcFeatureFlag> kSupportedServerRpcFeatureFlags = { APPLICATION_FEATURE_FLAGS };
+set<RpcFeatureFlag> kSupportedClientRpcFeatureFlags = { APPLICATION_FEATURE_FLAGS };
} // namespace rpc
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/inbound_call.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/inbound_call.cc b/src/kudu/rpc/inbound_call.cc
index 6ba184d..9fdaadf 100644
--- a/src/kudu/rpc/inbound_call.cc
+++ b/src/kudu/rpc/inbound_call.cc
@@ -30,9 +30,9 @@
#include "kudu/util/trace.h"
using google::protobuf::FieldDescriptor;
+using google::protobuf::io::CodedOutputStream;
using google::protobuf::Message;
using google::protobuf::MessageLite;
-using google::protobuf::io::CodedOutputStream;
using std::shared_ptr;
using std::vector;
using strings::Substitute;
@@ -42,7 +42,6 @@ DEFINE_bool(rpc_dump_all_traces, false,
TAG_FLAG(rpc_dump_all_traces, advanced);
TAG_FLAG(rpc_dump_all_traces, runtime);
-
namespace kudu {
namespace rpc {
@@ -80,6 +79,18 @@ void InboundCall::RespondSuccess(const MessageLite& response) {
Respond(response, true);
}
+void InboundCall::RespondUnsupportedFeature(const vector<uint32_t>& unsupported_features) {
+ TRACE_EVENT0("rpc", "InboundCall::RespondUnsupportedFeature");
+ ErrorStatusPB err;
+ err.set_message("unsupported feature flags");
+ err.set_code(ErrorStatusPB::ERROR_INVALID_REQUEST);
+ for (uint32_t feature : unsupported_features) {
+ err.add_unsupported_feature_flags(feature);
+ }
+
+ Respond(err, false);
+}
+
void InboundCall::RespondFailure(ErrorStatusPB::RpcErrorCodePB error_code,
const Status& status) {
TRACE_EVENT0("rpc", "InboundCall::RespondFailure");
@@ -274,5 +285,13 @@ MonoTime InboundCall::GetClientDeadline() const {
return deadline;
}
+vector<uint32_t> InboundCall::GetRequiredFeatures() const {
+ vector<uint32_t> features;
+ for (uint32_t feature : header_.required_feature_flags()) {
+ features.push_back(feature);
+ }
+ return features;
+}
+
} // namespace rpc
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/inbound_call.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/inbound_call.h b/src/kudu/rpc/inbound_call.h
index 88706d3..e15d323 100644
--- a/src/kudu/rpc/inbound_call.h
+++ b/src/kudu/rpc/inbound_call.h
@@ -102,6 +102,8 @@ class InboundCall {
void RespondFailure(ErrorStatusPB::RpcErrorCodePB error_code,
const Status &status);
+ void RespondUnsupportedFeature(const std::vector<uint32_t>& unsupported_features);
+
void RespondApplicationError(int error_ext_id, const std::string& message,
const google::protobuf::MessageLite& app_error_pb);
@@ -157,6 +159,10 @@ class InboundCall {
// If the client did not specify a deadline, returns MonoTime::Max().
MonoTime GetClientDeadline() const;
+ // Returns the set of application-specific feature flags required to service
+ // the RPC.
+ std::vector<uint32_t> GetRequiredFeatures() const;
+
private:
// Serialize and queue the response.
void Respond(const google::protobuf::MessageLite& response,
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/outbound_call.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/outbound_call.cc b/src/kudu/rpc/outbound_call.cc
index b63e69a..5b7fbd8 100644
--- a/src/kudu/rpc/outbound_call.cc
+++ b/src/kudu/rpc/outbound_call.cc
@@ -16,10 +16,12 @@
// under the License.
#include <algorithm>
-#include <string>
-#include <vector>
#include <boost/functional/hash.hpp>
#include <gflags/gflags.h>
+#include <set>
+#include <string>
+#include <unordered_set>
+#include <vector>
#include "kudu/gutil/strings/substitute.h"
#include "kudu/gutil/walltime.h"
@@ -35,9 +37,10 @@
namespace kudu {
namespace rpc {
-using strings::Substitute;
-using google::protobuf::Message;
using google::protobuf::io::CodedOutputStream;
+using google::protobuf::Message;
+using std::set;
+using strings::Substitute;
static const double kMicrosPerSecond = 1000000.0;
@@ -89,6 +92,10 @@ Status OutboundCall::SerializeTo(vector<Slice>* slices) {
header_.set_timeout_millis(timeout.ToMilliseconds());
}
+ for (uint32_t feature : controller_->required_server_features()) {
+ header_.add_required_feature_flags(feature);
+ }
+
CHECK_OK(serialization::SerializeHeader(header_, param_len, &header_buf_));
// Return the concatenated packet.
@@ -97,6 +104,14 @@ Status OutboundCall::SerializeTo(vector<Slice>* slices) {
return Status::OK();
}
+set<RpcFeatureFlag> OutboundCall::RequiredRpcFeatures() const {
+ set<RpcFeatureFlag> s;
+ if (!controller_->required_server_features().empty()) {
+ s.insert(RpcFeatureFlag::APPLICATION_FEATURE_FLAGS);
+ }
+ return s;
+}
+
Status OutboundCall::SetRequestParam(const Message& message) {
return serialization::SerializeMessage(message, &request_buf_);
}
@@ -111,7 +126,6 @@ const ErrorStatusPB* OutboundCall::error_pb() const {
return error_pb_.get();
}
-
string OutboundCall::StateName(State state) {
switch (state) {
case READY:
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/outbound_call.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/outbound_call.h b/src/kudu/rpc/outbound_call.h
index 2155752..3e65e52 100644
--- a/src/kudu/rpc/outbound_call.h
+++ b/src/kudu/rpc/outbound_call.h
@@ -17,6 +17,7 @@
#ifndef KUDU_RPC_CLIENT_CALL_H
#define KUDU_RPC_CLIENT_CALL_H
+#include <set>
#include <string>
#include <vector>
@@ -205,6 +206,8 @@ class OutboundCall {
// Fill in the call response.
void SetResponse(gscoped_ptr<CallResponse> resp);
+ std::set<RpcFeatureFlag> RequiredRpcFeatures() const;
+
std::string ToString() const;
void DumpPB(const DumpRunningRpcsRequestPB& req, RpcCallInProgressPB* resp);
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/rpc-test-base.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc-test-base.h b/src/kudu/rpc/rpc-test-base.h
index 0e0df7d..c629720 100644
--- a/src/kudu/rpc/rpc-test-base.h
+++ b/src/kudu/rpc/rpc-test-base.h
@@ -44,14 +44,15 @@
namespace kudu { namespace rpc {
-using kudu::rpc_test::AddRequestPartialPB;
using kudu::rpc_test::AddRequestPB;
+using kudu::rpc_test::AddRequestPartialPB;
using kudu::rpc_test::AddResponsePB;
using kudu::rpc_test::CalculatorError;
using kudu::rpc_test::CalculatorServiceIf;
using kudu::rpc_test::CalculatorServiceProxy;
using kudu::rpc_test::EchoRequestPB;
using kudu::rpc_test::EchoResponsePB;
+using kudu::rpc_test::FeatureFlags;
using kudu::rpc_test::PanicRequestPB;
using kudu::rpc_test::PanicResponsePB;
using kudu::rpc_test::SendTwoStringsRequestPB;
@@ -83,7 +84,7 @@ class GenericCalculatorService : public ServiceIf {
// this test doesn't generate metrics, so we ignore the argument.
}
- virtual void Handle(InboundCall *incoming) OVERRIDE {
+ void Handle(InboundCall *incoming) override {
if (incoming->remote_method().method_name() == kAddMethodName) {
DoAdd(incoming);
} else if (incoming->remote_method().method_name() == kSleepMethodName) {
@@ -96,7 +97,7 @@ class GenericCalculatorService : public ServiceIf {
}
}
- std::string service_name() const OVERRIDE { return kFullServiceName; }
+ std::string service_name() const override { return kFullServiceName; }
static std::string static_service_name() { return kFullServiceName; }
private:
@@ -164,16 +165,12 @@ class CalculatorService : public CalculatorServiceIf {
: CalculatorServiceIf(entity) {
}
- virtual void Add(const AddRequestPB *req,
- AddResponsePB *resp,
- RpcContext *context) OVERRIDE {
+ void Add(const AddRequestPB *req, AddResponsePB *resp, RpcContext *context) override {
resp->set_result(req->x() + req->y());
context->RespondSuccess();
}
- virtual void Sleep(const SleepRequestPB *req,
- SleepResponsePB *resp,
- RpcContext *context) OVERRIDE {
+ void Sleep(const SleepRequestPB *req, SleepResponsePB *resp, RpcContext *context) override {
if (req->return_app_error()) {
CalculatorError my_error;
my_error.set_extra_error_data("some application-specific error data");
@@ -206,16 +203,12 @@ class CalculatorService : public CalculatorServiceIf {
DoSleep(req, context);
}
- virtual void Echo(const EchoRequestPB *req,
- EchoResponsePB *resp,
- RpcContext *context) OVERRIDE {
+ void Echo(const EchoRequestPB *req, EchoResponsePB *resp, RpcContext *context) override {
resp->set_data(req->data());
context->RespondSuccess();
}
- virtual void WhoAmI(const WhoAmIRequestPB* req,
- WhoAmIResponsePB* resp,
- RpcContext* context) OVERRIDE {
+ void WhoAmI(const WhoAmIRequestPB* req, WhoAmIResponsePB* resp, RpcContext* context) override {
const UserCredentials& creds = context->user_credentials();
if (creds.has_effective_user()) {
resp->mutable_credentials()->set_effective_user(creds.effective_user());
@@ -225,19 +218,21 @@ class CalculatorService : public CalculatorServiceIf {
context->RespondSuccess();
}
- virtual void TestArgumentsInDiffPackage(const ReqDiffPackagePB *req,
- RespDiffPackagePB *resp,
- ::kudu::rpc::RpcContext *context) OVERRIDE {
+ void TestArgumentsInDiffPackage(const ReqDiffPackagePB *req,
+ RespDiffPackagePB *resp,
+ ::kudu::rpc::RpcContext *context) override {
context->RespondSuccess();
}
- virtual void Panic(const PanicRequestPB* req,
- PanicResponsePB* resp,
- RpcContext* context) OVERRIDE {
+ void Panic(const PanicRequestPB* req, PanicResponsePB* resp, RpcContext* context) override {
TRACE("Got panic request");
PANIC_RPC(context, "Test method panicking!");
}
+ bool SupportsFeature(uint32_t feature) const override {
+ return feature == FeatureFlags::FOO;
+ }
+
private:
void DoSleep(const SleepRequestPB *req,
RpcContext *context) {
@@ -266,11 +261,11 @@ class RpcTestBase : public KuduTest {
metric_entity_(METRIC_ENTITY_server.Instantiate(&metric_registry_, "test.rpc_test")) {
}
- virtual void SetUp() OVERRIDE {
+ void SetUp() override {
KuduTest::SetUp();
}
- virtual void TearDown() OVERRIDE {
+ void TearDown() override {
if (service_pool_) {
server_messenger_->UnregisterService(service_name_);
service_pool_->Shutdown();
@@ -421,7 +416,6 @@ class RpcTestBase : public KuduTest {
scoped_refptr<MetricEntity> metric_entity_;
};
-
} // namespace rpc
} // namespace kudu
#endif
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/rpc-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc-test.cc b/src/kudu/rpc/rpc-test.cc
index 03928af..e1d2c5b 100644
--- a/src/kudu/rpc/rpc-test.cc
+++ b/src/kudu/rpc/rpc-test.cc
@@ -26,9 +26,11 @@
#include "kudu/gutil/map-util.h"
#include "kudu/gutil/strings/join.h"
+#include "kudu/rpc/constants.h"
#include "kudu/rpc/serialization.h"
#include "kudu/util/countdown_latch.h"
#include "kudu/util/env.h"
+#include "kudu/util/scoped_cleanup.h"
#include "kudu/util/test_util.h"
METRIC_DECLARE_histogram(handler_latency_kudu_rpc_test_CalculatorService_Sleep);
@@ -511,5 +513,80 @@ TEST_F(TestRpc, TestRpcContextClientDeadline) {
ASSERT_OK(p.SyncRequest("Sleep", req, &resp, &controller));
}
+// Test that setting an call-level application feature flag to an unknown value
+// will make the server reject the call.
+TEST_F(TestRpc, TestApplicationFeatureFlag) {
+ // Set up server.
+ Sockaddr server_addr;
+ StartTestServerWithGeneratedCode(&server_addr);
+
+ // Set up client.
+ shared_ptr<Messenger> client_messenger(CreateMessenger("Client"));
+ Proxy p(client_messenger, server_addr, CalculatorService::static_service_name());
+
+ { // Supported flag
+ AddRequestPB req;
+ req.set_x(1);
+ req.set_y(2);
+ AddResponsePB resp;
+ RpcController controller;
+ controller.RequireServerFeature(FeatureFlags::FOO);
+ Status s = p.SyncRequest("Add", req, &resp, &controller);
+ SCOPED_TRACE(strings::Substitute("supported response: $0", s.ToString()));
+ ASSERT_TRUE(s.ok());
+ ASSERT_EQ(resp.result(), 3);
+ }
+
+ { // Unsupported flag
+ AddRequestPB req;
+ req.set_x(1);
+ req.set_y(2);
+ AddResponsePB resp;
+ RpcController controller;
+ controller.RequireServerFeature(FeatureFlags::FOO);
+ controller.RequireServerFeature(99);
+ Status s = p.SyncRequest("Add", req, &resp, &controller);
+ SCOPED_TRACE(strings::Substitute("unsupported response: $0", s.ToString()));
+ ASSERT_TRUE(s.IsRemoteError());
+ }
+}
+
+TEST_F(TestRpc, TestApplicationFeatureFlagUnsupportedServer) {
+ auto savedFlags = kSupportedServerRpcFeatureFlags;
+ auto cleanup = MakeScopedCleanup([&] () { kSupportedServerRpcFeatureFlags = savedFlags; });
+ kSupportedServerRpcFeatureFlags = {};
+
+ // Set up server.
+ Sockaddr server_addr;
+ StartTestServerWithGeneratedCode(&server_addr);
+
+ // Set up client.
+ shared_ptr<Messenger> client_messenger(CreateMessenger("Client"));
+ Proxy p(client_messenger, server_addr, CalculatorService::static_service_name());
+
+ { // Required flag
+ AddRequestPB req;
+ req.set_x(1);
+ req.set_y(2);
+ AddResponsePB resp;
+ RpcController controller;
+ controller.RequireServerFeature(FeatureFlags::FOO);
+ Status s = p.SyncRequest("Add", req, &resp, &controller);
+ SCOPED_TRACE(strings::Substitute("supported response: $0", s.ToString()));
+ ASSERT_TRUE(s.IsNotSupported());
+ }
+
+ { // No required flag
+ AddRequestPB req;
+ req.set_x(1);
+ req.set_y(2);
+ AddResponsePB resp;
+ RpcController controller;
+ Status s = p.SyncRequest("Add", req, &resp, &controller);
+ SCOPED_TRACE(strings::Substitute("supported response: $0", s.ToString()));
+ ASSERT_TRUE(s.ok());
+ }
+}
+
} // namespace rpc
} // namespace kudu
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/rpc_controller.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc_controller.cc b/src/kudu/rpc/rpc_controller.cc
index 53f76d9..7236b0a 100644
--- a/src/kudu/rpc/rpc_controller.cc
+++ b/src/kudu/rpc/rpc_controller.cc
@@ -87,6 +87,11 @@ void RpcController::set_deadline(const MonoTime& deadline) {
set_timeout(deadline.GetDeltaSince(MonoTime::Now(MonoTime::FINE)));
}
+void RpcController::RequireServerFeature(uint32_t feature) {
+ DCHECK(!call_ || call_->state() == OutboundCall::READY);
+ required_server_features_.insert(feature);
+}
+
MonoDelta RpcController::timeout() const {
lock_guard<simple_spinlock> l(&lock_);
return timeout_;
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/rpc_controller.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc_controller.h b/src/kudu/rpc/rpc_controller.h
index bea703b..5abafc2 100644
--- a/src/kudu/rpc/rpc_controller.h
+++ b/src/kudu/rpc/rpc_controller.h
@@ -17,8 +17,10 @@
#ifndef KUDU_RPC_RPC_CONTROLLER_H
#define KUDU_RPC_RPC_CONTROLLER_H
+#include <functional>
#include <glog/logging.h>
#include <memory>
+#include <unordered_set>
#include "kudu/gutil/macros.h"
#include "kudu/util/locks.h"
@@ -102,6 +104,54 @@ class RpcController {
// Using an uninitialized deadline means the call won't time out.
void set_deadline(const MonoTime& deadline);
+ // Add a requirement that the server side must support a feature with the
+ // given identifier. The set of required features is sent to the server
+ // with the RPC call, and if any required feature is not supported, the
+ // call will fail with a NotSupported() status.
+ //
+ // This can be used when an RPC call changes in a way that is protobuf-compatible,
+ // but for which it would not be appropriate for the server to simply ignore
+ // an added field. For example, consider an API call like:
+ //
+ // message DeleteAccount {
+ // optional string username = 1;
+ // optional bool dry_run = 2; // ADDED LATER!
+ // }
+ //
+ // In this case, if a new client which supports the 'dry_run' flag sends the RPC
+ // to an old server, the old server will simply ignore the unrecognized parameter,
+ // with highly problematic results. To solve this problem, the new version can
+ // add a feature flag:
+ //
+ // In .proto file
+ // ----------------
+ // enum MyFeatureFlags {
+ // UNKNOWN = 0;
+ // DELETE_ACCOUNT_SUPPORTS_DRY_RUN = 1;
+ // }
+ //
+ // In client code:
+ // ---------------
+ // if (dry_run) {
+ // rpc.RequireServerFeature(DELETE_ACCOUNT_SUPPORTS_DRY_RUN);
+ // req.set_dry_run(true);
+ // }
+ //
+ // This has the effect of (a) maintaining compatibility when dry_run is not specified
+ // and (b) rejecting the RPC with a "NotSupported" error when it is.
+ //
+ // NOTE: 'feature' is an int rather than an enum type because each service
+ // must define its own enum of supported features, and protobuf doesn't support
+ // any ability to 'extend' enum types. Implementers should define an enum in the
+ // service's protobuf definition as shown above.
+ void RequireServerFeature(uint32_t feature);
+
+ // Executes the provided function with a reference to the required server
+ // features.
+ const std::unordered_set<uint32_t>& required_server_features() const {
+ return required_server_features_;
+ }
+
// Return the configured timeout.
MonoDelta timeout() const;
@@ -119,6 +169,7 @@ class RpcController {
friend class Proxy;
MonoDelta timeout_;
+ std::unordered_set<uint32_t> required_server_features_;
mutable simple_spinlock lock_;
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/rpc_header.proto
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc_header.proto b/src/kudu/rpc/rpc_header.proto
index 100a57d..b1fabc3 100644
--- a/src/kudu/rpc/rpc_header.proto
+++ b/src/kudu/rpc/rpc_header.proto
@@ -21,7 +21,6 @@ package kudu.rpc;
option java_package = "org.kududb.rpc";
-
// The Kudu RPC protocol is similar to the RPC protocol of Hadoop and HBase.
// See the following for reference on those other protocols:
// - https://issues.apache.org/jira/browse/HBASE-7898
@@ -61,9 +60,9 @@ message ConnectionContextPB {
enum RpcFeatureFlag {
UNKNOWN = 0;
- // TODO: as soon as we have any real feature flag, we can get rid of this
- // feature flag. For now, we just need something to assert on for tests.
- TMP_TEST_FEATURE_FLAG = 1000;
+ // The RPC system is required to support application feature flags in the
+ // request and response headers.
+ APPLICATION_FEATURE_FLAGS = 1;
};
// Message type passed back & forth for the SASL negotiation.
@@ -130,6 +129,17 @@ message RequestHeader {
// transit time between the client and server, if you wait exactly this amount of
// time and then respond, you are likely to cause a timeout on the client.
optional uint32 timeout_millis = 10;
+
+ // Feature flags that the service must support in order to properly interpret this
+ // request. The client can pass any set of flags, and if the server doesn't
+ // support any of them, then it will fail the request.
+ //
+ // NOTE: these are for evolving features at the level of the application, not
+ // the RPC framework. Hence, we have to use a generic int type rather than a
+ // particular enum.
+ // NOTE: the server will only interpret this field if it supports the
+ // APPLICATION_FEATURE_FLAGS flag.
+ repeated uint32 required_feature_flags = 11;
}
message ResponseHeader {
@@ -169,7 +179,8 @@ message ErrorStatusPB {
// The server is overloaded - the client should try again shortly.
ERROR_SERVER_TOO_BUSY = 4;
- // The request parameter was not parseable or was missing required fields.
+ // The request parameter was not parseable, was missing required fields,
+ // or the server does not support the required feature flags.
ERROR_INVALID_REQUEST = 5;
// FATAL_* errors indicate that the client should shut down the connection.
@@ -191,6 +202,10 @@ message ErrorStatusPB {
// TODO: Make code required?
optional RpcErrorCodePB code = 2; // Specific error identifier.
+ // If the request is failed due to an unsupported feature flag, the particular
+ // flag(s) that were not supported will be sent back to the client.
+ repeated uint32 unsupported_feature_flags = 3;
+
// Allow extensions. When the RPC returns ERROR_APPLICATION, the server
// should also fill in exactly one of these extension fields, which contains
// more details on the service-specific error.
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/rtest.proto
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rtest.proto b/src/kudu/rpc/rtest.proto
index f1e3e93..2c89877 100644
--- a/src/kudu/rpc/rtest.proto
+++ b/src/kudu/rpc/rtest.proto
@@ -90,6 +90,11 @@ message CalculatorError {
message PanicRequestPB {}
message PanicResponsePB {}
+enum FeatureFlags {
+ UNKNOWN=0;
+ FOO=1;
+}
+
service CalculatorService {
rpc Add(AddRequestPB) returns(AddResponsePB);
rpc Sleep(SleepRequestPB) returns(SleepResponsePB);
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/sasl_rpc-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/sasl_rpc-test.cc b/src/kudu/rpc/sasl_rpc-test.cc
index 25ea80d..1a2d2e8 100644
--- a/src/kudu/rpc/sasl_rpc-test.cc
+++ b/src/kudu/rpc/sasl_rpc-test.cc
@@ -118,7 +118,7 @@ static void RunPlainNegotiationServer(Socket* conn) {
CHECK_OK(sasl_server.Init(kSaslAppName));
CHECK_OK(sasl_server.EnablePlain(std::move(authstore)));
CHECK_OK(sasl_server.Negotiate());
- CHECK(ContainsKey(sasl_server.client_features(), TMP_TEST_FEATURE_FLAG));
+ CHECK(ContainsKey(sasl_server.client_features(), APPLICATION_FEATURE_FLAGS));
}
static void RunPlainNegotiationClient(Socket* conn) {
@@ -126,7 +126,7 @@ static void RunPlainNegotiationClient(Socket* conn) {
CHECK_OK(sasl_client.Init(kSaslAppName));
CHECK_OK(sasl_client.EnablePlain("danger", "burrito"));
CHECK_OK(sasl_client.Negotiate());
- CHECK(ContainsKey(sasl_client.server_features(), TMP_TEST_FEATURE_FLAG));
+ CHECK(ContainsKey(sasl_client.server_features(), APPLICATION_FEATURE_FLAGS));
}
// Test SASL negotiation using the PLAIN mechanism over a socket.
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/service_if.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/service_if.cc b/src/kudu/rpc/service_if.cc
index 7377cea..bbc9a2d 100644
--- a/src/kudu/rpc/service_if.cc
+++ b/src/kudu/rpc/service_if.cc
@@ -44,6 +44,10 @@ ServiceIf::~ServiceIf() {
void ServiceIf::Shutdown() {
}
+bool ServiceIf::SupportsFeature(uint32_t feature) const {
+ return false;
+}
+
bool ServiceIf::ParseParam(InboundCall *call, google::protobuf::Message *message) {
Slice param(call->serialized_request());
if (PREDICT_FALSE(!message->ParseFromArray(param.data(), param.size()))) {
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/service_if.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/service_if.h b/src/kudu/rpc/service_if.h
index 0224420..83f49dd 100644
--- a/src/kudu/rpc/service_if.h
+++ b/src/kudu/rpc/service_if.h
@@ -53,6 +53,10 @@ class ServiceIf {
virtual void Shutdown();
virtual std::string service_name() const = 0;
+ // The service should return true if it supports the provided application
+ // specific feature flag.
+ virtual bool SupportsFeature(uint32_t feature) const;
+
protected:
bool ParseParam(InboundCall* call, google::protobuf::Message* message);
void RespondBadMethod(InboundCall* call);
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/service_pool.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/service_pool.cc b/src/kudu/rpc/service_pool.cc
index 55a2791..eb7d46a 100644
--- a/src/kudu/rpc/service_pool.cc
+++ b/src/kudu/rpc/service_pool.cc
@@ -106,6 +106,18 @@ void ServicePool::Shutdown() {
Status ServicePool::QueueInboundCall(gscoped_ptr<InboundCall> call) {
InboundCall* c = call.release();
+ vector<uint32_t> unsupported_features;
+ for (uint32_t feature : c->GetRequiredFeatures()) {
+ if (!service_->SupportsFeature(feature)) {
+ unsupported_features.push_back(feature);
+ }
+ }
+
+ if (!unsupported_features.empty()) {
+ c->RespondUnsupportedFeature(unsupported_features);
+ return Status::NotSupported("call requires unsupported application feature flags");
+ }
+
TRACE_TO(c->trace(), "Inserting onto call queue");
// Queue message on service queue
QueueStatus queue_status = service_queue_.Put(c);
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/transfer.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/transfer.cc b/src/kudu/rpc/transfer.cc
index b1807ae..553de5f 100644
--- a/src/kudu/rpc/transfer.cc
+++ b/src/kudu/rpc/transfer.cc
@@ -42,6 +42,7 @@ namespace kudu {
namespace rpc {
using std::ostringstream;
+using std::set;
using std::string;
#define RETURN_ON_ERROR_OR_SOCKET_NOT_READY(status) \
@@ -122,11 +123,15 @@ string InboundTransfer::StatusAsString() const {
return strings::Substitute("$0/$1 bytes received", cur_offset_, total_length_);
}
-OutboundTransfer::OutboundTransfer(const std::vector<Slice> &payload,
+OutboundTransfer::OutboundTransfer(int32_t call_id,
+ const std::vector<Slice> &payload,
+ set<RpcFeatureFlag> required_features,
TransferCallbacks *callbacks)
: cur_slice_idx_(0),
cur_offset_in_slice_(0),
+ required_features_(std::move(required_features)),
callbacks_(callbacks),
+ call_id_(call_id),
aborted_(false) {
CHECK(!payload.empty());
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/rpc/transfer.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/transfer.h b/src/kudu/rpc/transfer.h
index cc66b4e..662d3e4 100644
--- a/src/kudu/rpc/transfer.h
+++ b/src/kudu/rpc/transfer.h
@@ -18,10 +18,11 @@
#ifndef KUDU_RPC_TRANSFER_H
#define KUDU_RPC_TRANSFER_H
-#include <boost/intrusive/list.hpp>
#include <boost/function.hpp>
+#include <boost/intrusive/list.hpp>
#include <boost/utility.hpp>
#include <gflags/gflags.h>
+#include <set>
#include <stdint.h>
#include <string>
#include <vector>
@@ -87,7 +88,6 @@ class InboundTransfer {
DISALLOW_COPY_AND_ASSIGN(InboundTransfer);
};
-
// When the connection wants to send data, it creates an OutboundTransfer object
// to encompass it. This sits on a queue within the Connection, so that each time
// the Connection wakes up with a writable socket, it consumes more bytes off
@@ -106,9 +106,17 @@ class OutboundTransfer : public boost::intrusive::list_base_hook<> {
// memory of the slices. The slices must remain valid until the callback
// is triggered.
//
+ // When the transfer starts, the required features are checked against the set
+ // of features which the server supports. The check is delayed until just
+ // before the transfer starts because it depends on the negotiation with the
+ // server being complete. The call_id allows the call to be canceled if
+ // required features are not supported.
+ //
// NOTE: 'payload' is currently restricted to a maximum of kMaxPayloadSlices
// slices.
- OutboundTransfer(const std::vector<Slice> &payload,
+ OutboundTransfer(int32_t call_id,
+ const std::vector<Slice> &payload,
+ std::set<RpcFeatureFlag> required_features,
TransferCallbacks *callbacks);
// Destruct the transfer. A transfer object should never be deallocated
@@ -128,11 +136,19 @@ class OutboundTransfer : public boost::intrusive::list_base_hook<> {
// Return true if the entire transfer has been sent.
bool TransferFinished() const;
+ const std::set<RpcFeatureFlag>& required_features() const {
+ return required_features_;
+ }
+
// Return the total number of bytes to be sent (including those already sent)
int32_t TotalLength() const;
std::string HexDump() const;
+ int32_t call_id() const {
+ return call_id_;
+ }
+
private:
// Slices to send. Uses an array here instead of a vector to avoid an expensive
// vector construction (improved performance a couple percent).
@@ -144,8 +160,12 @@ class OutboundTransfer : public boost::intrusive::list_base_hook<> {
// The number of bytes in the above slice which has already been sent.
int32_t cur_offset_in_slice_;
+ std::set<RpcFeatureFlag> required_features_;
+
TransferCallbacks *callbacks_;
+ int32_t call_id_;
+
bool aborted_;
DISALLOW_COPY_AND_ASSIGN(OutboundTransfer);
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/util/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/util/CMakeLists.txt b/src/kudu/util/CMakeLists.txt
index db94115..241f6f6 100644
--- a/src/kudu/util/CMakeLists.txt
+++ b/src/kudu/util/CMakeLists.txt
@@ -279,10 +279,10 @@ ADD_KUDU_TEST(jsonreader-test)
ADD_KUDU_TEST(knapsack_solver-test)
ADD_KUDU_TEST(logging-test)
ADD_KUDU_TEST(map-util-test)
+ADD_KUDU_TEST(mem_tracker-test)
ADD_KUDU_TEST(memcmpable_varint-test LABELS no_tsan)
ADD_KUDU_TEST(memenv/memenv-test)
ADD_KUDU_TEST(memory/arena-test)
-ADD_KUDU_TEST(mem_tracker-test)
ADD_KUDU_TEST(metrics-test)
ADD_KUDU_TEST(monotime-test)
ADD_KUDU_TEST(mt-hdr_histogram-test RUN_SERIAL true)
@@ -303,6 +303,7 @@ ADD_KUDU_TEST(rolling_log-test)
ADD_KUDU_TEST(rw_semaphore-test)
ADD_KUDU_TEST(rwc_lock-test)
ADD_KUDU_TEST(safe_math-test)
+ADD_KUDU_TEST(scoped_cleanup-test)
ADD_KUDU_TEST(slice-test)
ADD_KUDU_TEST(spinlock_profiling-test)
ADD_KUDU_TEST(stack_watchdog-test)
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/util/scoped_cleanup-test.cc
----------------------------------------------------------------------
diff --git a/src/kudu/util/scoped_cleanup-test.cc b/src/kudu/util/scoped_cleanup-test.cc
new file mode 100644
index 0000000..790df39
--- /dev/null
+++ b/src/kudu/util/scoped_cleanup-test.cc
@@ -0,0 +1,34 @@
+// 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/util/scoped_cleanup.h"
+
+#include <gtest/gtest.h>
+
+namespace kudu {
+
+TEST(ScopedCleanup, TestCleanup) {
+ int var = 0;
+ {
+ auto saved = var;
+ auto cleanup = MakeScopedCleanup([&] () { var = saved; });
+ var = 42;
+ }
+ ASSERT_EQ(var, 0);
+}
+
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/incubator-kudu/blob/f82919ed/src/kudu/util/scoped_cleanup.h
----------------------------------------------------------------------
diff --git a/src/kudu/util/scoped_cleanup.h b/src/kudu/util/scoped_cleanup.h
new file mode 100644
index 0000000..a74422c
--- /dev/null
+++ b/src/kudu/util/scoped_cleanup.h
@@ -0,0 +1,40 @@
+// 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.
+
+#pragma once
+
+#include <utility>
+
+namespace kudu {
+
+// A scoped object which runs a cleanup function when going out of scope. Can
+// be used for scoped resource cleanup.
+template<typename F>
+class ScopedCleanup {
+ public:
+ explicit ScopedCleanup(F f) : f_(std::move(f)) {}
+ ~ScopedCleanup() { f_(); }
+ private:
+ F f_;
+};
+
+// Creates a new scoped cleanup instance with the provided function.
+template<typename F>
+ScopedCleanup<F> MakeScopedCleanup(F f) {
+ return ScopedCleanup<F>(f);
+}
+} // namespace kudu