You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by al...@apache.org on 2017/05/17 01:40:52 UTC
[2/2] kudu git commit: [c++ client] re-acquire authn token if expired
[c++ client] re-acquire authn token if expired
Updated C++ client to re-acquire authn token if server responds with
FATAL_INVALID_AUTHENTICATION_TOKEN error on connection negotiation.
If that happens while negotiating a connection for a call, the
connection is closed and the client reconnects to the cluster to get
a new authn token. After successfully re-acquiring authn token the
client retries the RPC call again.
Added corresponding integration test to cover authn token re-acquisition
for various scenarios.
Change-Id: I418497ac59cfd4e476e9bfc6abe6b10b487712f5
Reviewed-on: http://gerrit.cloudera.org:8080/6648
Tested-by: Kudu Jenkins
Reviewed-by: Dan Burkert <da...@apache.org>
Reviewed-by: Todd Lipcon <to...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/kudu/repo
Commit: http://git-wip-us.apache.org/repos/asf/kudu/commit/3db551f7
Tree: http://git-wip-us.apache.org/repos/asf/kudu/tree/3db551f7
Diff: http://git-wip-us.apache.org/repos/asf/kudu/diff/3db551f7
Branch: refs/heads/master
Commit: 3db551f7b31def26c0fa04184843d3dfa6161d9a
Parents: 80700aa
Author: Alexey Serbin <as...@cloudera.com>
Authored: Fri Apr 14 15:34:16 2017 -0700
Committer: Alexey Serbin <as...@cloudera.com>
Committed: Wed May 17 01:35:59 2017 +0000
----------------------------------------------------------------------
docs/known_issues.adoc | 2 +-
docs/security.adoc | 10 +-
src/kudu/client/batcher.cc | 24 ++
src/kudu/client/client-internal.cc | 113 +++++--
src/kudu/client/client-internal.h | 24 +-
src/kudu/client/master_rpc.cc | 41 ++-
src/kudu/client/master_rpc.h | 5 +-
src/kudu/client/meta_cache.cc | 38 ++-
src/kudu/client/scanner-internal.cc | 35 +-
src/kudu/client/scanner-internal.h | 4 +
src/kudu/integration-tests/CMakeLists.txt | 1 +
.../authn_token_expire-itest.cc | 320 +++++++++++++++++++
src/kudu/integration-tests/cluster_verifier.h | 4 +-
src/kudu/rpc/client_negotiation.cc | 48 +--
src/kudu/rpc/client_negotiation.h | 15 +-
src/kudu/rpc/connection.cc | 31 +-
src/kudu/rpc/connection.h | 7 +-
src/kudu/rpc/negotiation.cc | 15 +-
src/kudu/rpc/outbound_call.cc | 8 +-
src/kudu/rpc/reactor.cc | 15 +-
src/kudu/rpc/reactor.h | 6 +-
src/kudu/rpc/retriable_rpc.h | 80 +++--
src/kudu/rpc/rpc.h | 5 +
src/kudu/rpc/server_negotiation.cc | 39 +++
24 files changed, 751 insertions(+), 139 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/docs/known_issues.adoc
----------------------------------------------------------------------
diff --git a/docs/known_issues.adoc b/docs/known_issues.adoc
index 8828f4b..1ff9865 100644
--- a/docs/known_issues.adoc
+++ b/docs/known_issues.adoc
@@ -169,7 +169,7 @@
* Kudu client instances retrieve authentication tokens upon first contact with the
cluster. These tokens expire after one week. Use of a single Kudu client instance
- for more than one week is not supported.
+ for more than one week is only supported by the C++ client but not by the Java client.
== Other Known Issues
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/docs/security.adoc
----------------------------------------------------------------------
diff --git a/docs/security.adoc b/docs/security.adoc
index 4ad48c5..1b8cbf7 100644
--- a/docs/security.adoc
+++ b/docs/security.adoc
@@ -218,11 +218,11 @@ Kudu has a few known security limitations:
// TODO(danburkert): add JIRA links for each of these.
-Long-lived Tokens:: Kudu clients do not automatically request fresh tokens after
-initial token expiration, so long-lived clients in secure clusters are not
-supported. Note that applications such as Apache Impala construct new clients
-for each query and thus this limitation only affects the runtime of any single
-query.
+Long-lived Tokens:: The Java Kudu client does not automatically request fresh
+authn tokens after initial token expiration, so long-lived Java clients in
+secure clusters are not supported. However, the C++ Kudu client does
+automatically request fresh authn tokens, so long-lived C++ clients
+(i.e. beyond authn token lifetime) in secure clusters are supported.
Custom Kerberos Principal:: Kudu does not support setting a custom service
principal for Kudu processes. The principal must be 'kudu'.
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/client/batcher.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/batcher.cc b/src/kudu/client/batcher.cc
index 1586cae..498bebe 100644
--- a/src/kudu/client/batcher.cc
+++ b/src/kudu/client/batcher.cc
@@ -49,6 +49,7 @@
#include "kudu/rpc/request_tracker.h"
#include "kudu/rpc/retriable_rpc.h"
#include "kudu/rpc/rpc.h"
+#include "kudu/rpc/rpc_controller.h"
#include "kudu/rpc/rpc_header.pb.h"
#include "kudu/tserver/tserver_service.proxy.h"
#include "kudu/util/debug-util.h"
@@ -64,6 +65,7 @@ using strings::Substitute;
namespace kudu {
+using rpc::CredentialsPolicy;
using rpc::ErrorStatusPB;
using rpc::Messenger;
using rpc::RequestTracker;
@@ -219,6 +221,7 @@ class WriteRpc : public RetriableRpc<RemoteTabletServer, WriteRequestPB, WriteRe
void Try(RemoteTabletServer* replica, const ResponseCallback& callback) override;
RetriableRpcStatus AnalyzeResponse(const Status& rpc_cb_status) override;
void Finish(const Status& status) override;
+ bool GetNewAuthnTokenAndRetry() override;
private:
// Pointer back to the batcher. Processes the write response when it
@@ -354,6 +357,17 @@ RetriableRpcStatus WriteRpc::AnalyzeResponse(const Status& rpc_cb_status) {
return result;
}
+ // Check whether it's an invalid authn token. That's the error code the server
+ // sends back if authn token is expired.
+ if (result.status.IsNotAuthorized()) {
+ const ErrorStatusPB* err = mutable_retrier()->controller().error_response();
+ if (err && err->has_code() &&
+ err->code() == ErrorStatusPB::FATAL_INVALID_AUTHENTICATION_TOKEN) {
+ result.result = RetriableRpcStatus::INVALID_AUTHENTICATION_TOKEN;
+ return result;
+ }
+ }
+
// Failover to a replica in the event of any network failure or of a DNS resolution problem.
//
// TODO(adar): This is probably too harsh; some network failures should be
@@ -396,6 +410,16 @@ RetriableRpcStatus WriteRpc::AnalyzeResponse(const Status& rpc_cb_status) {
return result;
}
+bool WriteRpc::GetNewAuthnTokenAndRetry() {
+ // To get a new authn token it's necessary to authenticate with the master
+ // using any other credentials but already existing authn token.
+ KuduClient* c = batcher_->client_;
+ c->data_->ConnectToClusterAsync(c, retrier().deadline(),
+ Bind(&RetriableRpc::GetNewAuthnTokenAndRetryCb, Unretained(this)),
+ CredentialsPolicy::PRIMARY_CREDENTIALS);
+ return true;
+}
+
Batcher::Batcher(KuduClient* client,
scoped_refptr<ErrorCollector> error_collector,
sp::weak_ptr<KuduSession> session,
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/client/client-internal.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/client-internal.cc b/src/kudu/client/client-internal.cc
index 3f45d02..30723d2 100644
--- a/src/kudu/client/client-internal.cc
+++ b/src/kudu/client/client-internal.cc
@@ -82,8 +82,8 @@ using master::ListTabletServersResponsePB;
using master::MasterErrorPB;
using master::MasterFeatures;
using master::MasterServiceProxy;
+using rpc::CredentialsPolicy;
using rpc::ErrorStatusPB;
-using rpc::Rpc;
using rpc::RpcController;
using strings::Substitute;
@@ -211,6 +211,23 @@ Status KuduClient::Data::SyncLeaderMasterRpc(
}
}
+ if (s.IsNotAuthorized()) {
+ const ErrorStatusPB* err = rpc.error_response();
+ if (err && err->has_code() &&
+ err->code() == ErrorStatusPB::FATAL_INVALID_AUTHENTICATION_TOKEN) {
+ // Assuming the token has expired: it's necessary to get a new one.
+ LOG(INFO) << "Reconnecting to the cluster for a new authn token";
+ const Status connect_status = ConnectToCluster(
+ client, deadline, CredentialsPolicy::PRIMARY_CREDENTIALS);
+ if (PREDICT_FALSE(!connect_status.ok())) {
+ KLOG_EVERY_N_SECS(WARNING, 1)
+ << "Unable to reconnect to the cluster for a new authn token: "
+ << connect_status.ToString();
+ }
+ continue;
+ }
+ }
+
if (s.IsServiceUnavailable()) {
KLOG_EVERY_N_SECS(WARNING, 1)
<< "Unable to send the request (" << SecureShortDebugString(req)
@@ -609,7 +626,8 @@ Status KuduClient::Data::GetTableSchema(KuduClient* client,
void KuduClient::Data::ConnectedToClusterCb(
const Status& status,
const Sockaddr& leader_addr,
- const master::ConnectToMasterResponsePB& connect_response) {
+ const master::ConnectToMasterResponsePB& connect_response,
+ CredentialsPolicy cred_policy) {
// Ensure that all of the CAs reported by the master are trusted
// in our local TLS configuration.
@@ -636,13 +654,19 @@ void KuduClient::Data::ConnectedToClusterCb(
// Adopt the authentication token from the response, if it's been set.
if (connect_response.has_authn_token()) {
messenger_->set_authn_token(connect_response.authn_token());
+ LOG(INFO) << "Received and adopted authn token";
}
vector<StatusCallback> cbs;
{
std::lock_guard<simple_spinlock> l(leader_master_lock_);
- cbs.swap(leader_master_callbacks_);
- leader_master_rpc_.reset();
+ if (cred_policy == CredentialsPolicy::PRIMARY_CREDENTIALS) {
+ leader_master_rpc_primary_creds_.reset();
+ cbs.swap(leader_master_callbacks_primary_creds_);
+ } else {
+ leader_master_rpc_any_creds_.reset();
+ cbs.swap(leader_master_callbacks_any_creds_);
+ }
if (status.ok()) {
leader_master_hostport_ = HostPort(leader_addr);
@@ -656,15 +680,17 @@ void KuduClient::Data::ConnectedToClusterCb(
}
Status KuduClient::Data::ConnectToCluster(KuduClient* client,
- const MonoTime& deadline) {
+ const MonoTime& deadline,
+ CredentialsPolicy creds_policy) {
Synchronizer sync;
- ConnectToClusterAsync(client, deadline, sync.AsStatusCallback());
+ ConnectToClusterAsync(client, deadline, sync.AsStatusCallback(), creds_policy);
return sync.Wait();
}
void KuduClient::Data::ConnectToClusterAsync(KuduClient* client,
const MonoTime& deadline,
- const StatusCallback& cb) {
+ const StatusCallback& cb,
+ CredentialsPolicy creds_policy) {
DCHECK(deadline.Initialized());
vector<Sockaddr> master_sockaddrs;
@@ -690,29 +716,68 @@ void KuduClient::Data::ConnectToClusterAsync(KuduClient* client,
master_sockaddrs.push_back(addrs[0]);
}
- // This ensures that no more than one ConnectToClusterRpc is in
- // flight at a time -- there isn't much sense in requesting this information
- // in parallel, since the requests should end up with the same result.
- // Instead, we simply piggy-back onto the existing request by adding our own
- // callback to leader_master_callbacks_.
+ // This ensures that no more than one ConnectToClusterRpc of each credentials
+ // policy is in flight at any time -- there isn't much sense in requesting
+ // this information in parallel, since the requests should end up with the
+ // same result. Instead, simply piggy-back onto the existing request by adding
+ // our the callback to leader_master_callbacks_{any_creds,primary_creds}_.
std::unique_lock<simple_spinlock> l(leader_master_lock_);
- leader_master_callbacks_.push_back(cb);
- if (!leader_master_rpc_) {
- // No one is sending a request yet - we need to be the one to do it.
- leader_master_rpc_.reset(new internal::ConnectToClusterRpc(
- std::bind(&KuduClient::Data::ConnectedToClusterCb, this,
- std::placeholders::_1,
- std::placeholders::_2,
- std::placeholders::_3),
+
+ // Optimize sending out a new request in the presence of already existing
+ // requests to the leader master. Depending on the credentials policy for the
+ // request, we select different strategies:
+ // * If the new request is of ANY_CREDENTIALS policy, piggy-back it on any
+ // leader master request which is progress.
+ // * If the new request is of PRIMARY_CREDENTIALS, it can by piggy-backed
+ // only on the request of the same type.
+ //
+ // If this is a request to re-acquire authn token, we allow it to run in
+ // parallel with other requests of ANY_CREDENTIALS_TYPE policy, if any.
+ // Otherwise it's hard to guarantee that the token re-acquisition request
+ // will be sent: there might be other concurrent requests to leader master,
+ // they might have different timeout settings and they are retried
+ // independently of each other.
+ DCHECK(creds_policy == CredentialsPolicy::ANY_CREDENTIALS ||
+ creds_policy == CredentialsPolicy::PRIMARY_CREDENTIALS);
+
+ // Decide on which call to piggy-back the callback: if a call with primary
+ // credentials is available, piggy-back on that. Otherwise, piggy-back on
+ // any-credentials-policy request.
+ if (leader_master_rpc_primary_creds_) {
+ // Piggy-back on an existing connection with primary credentials.
+ leader_master_callbacks_primary_creds_.push_back(cb);
+ } else if (leader_master_rpc_any_creds_ &&
+ creds_policy == CredentialsPolicy::ANY_CREDENTIALS) {
+ // Piggy-back on an existing connection with any credentials.
+ leader_master_callbacks_any_creds_.push_back(cb);
+ } else {
+ // It's time to create a new request which would satisfy the credentials
+ // policy.
+ scoped_refptr<internal::ConnectToClusterRpc> rpc(
+ new internal::ConnectToClusterRpc(
+ std::bind(&KuduClient::Data::ConnectedToClusterCb, this,
+ std::placeholders::_1,
+ std::placeholders::_2,
+ std::placeholders::_3,
+ creds_policy),
std::move(master_sockaddrs),
deadline,
client->default_rpc_timeout(),
- messenger_));
+ messenger_,
+ creds_policy));
+
+ if (creds_policy == CredentialsPolicy::PRIMARY_CREDENTIALS) {
+ DCHECK(!leader_master_rpc_primary_creds_);
+ leader_master_rpc_primary_creds_ = rpc;
+ leader_master_callbacks_primary_creds_.push_back(cb);
+ } else {
+ DCHECK(!leader_master_rpc_any_creds_);
+ leader_master_rpc_any_creds_ = rpc;
+ leader_master_callbacks_any_creds_.push_back(cb);
+ }
l.unlock();
- leader_master_rpc_->SendRpc();
+ rpc->SendRpc();
}
-
-
}
HostPort KuduClient::Data::leader_master_hostport() const {
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/client/client-internal.h
----------------------------------------------------------------------
diff --git a/src/kudu/client/client-internal.h b/src/kudu/client/client-internal.h
index c211639..a7a1434 100644
--- a/src/kudu/client/client-internal.h
+++ b/src/kudu/client/client-internal.h
@@ -25,9 +25,9 @@
#include <vector>
#include <boost/function.hpp>
-#include <boost/optional.hpp>
#include "kudu/client/client.h"
+#include "kudu/rpc/rpc_controller.h"
#include "kudu/security/token.pb.h"
#include "kudu/util/atomic.h"
#include "kudu/util/locks.h"
@@ -49,7 +49,6 @@ class MasterServiceProxy;
namespace rpc {
class Messenger;
class RequestTracker;
-class RpcController;
} // namespace rpc
namespace client {
@@ -141,7 +140,8 @@ class KuduClient::Data {
// See also: ConnectToClusterAsync.
void ConnectedToClusterCb(const Status& status,
const Sockaddr& leader_addr,
- const master::ConnectToMasterResponsePB& connect_response);
+ const master::ConnectToMasterResponsePB& connect_response,
+ rpc::CredentialsPolicy cred_policy);
// Asynchronously sets 'master_proxy_' to the leader master by
// cycling through servers listed in 'master_server_addrs_' until
@@ -153,13 +153,17 @@ class KuduClient::Data {
// Works with both a distributed and non-distributed configuration.
void ConnectToClusterAsync(KuduClient* client,
const MonoTime& deadline,
- const StatusCallback& cb);
+ const StatusCallback& cb,
+ rpc::CredentialsPolicy creds_policy);
// Synchronous version of ConnectToClusterAsync method above.
//
// NOTE: since this uses a Synchronizer, this may not be invoked by
// a method that's on a reactor thread.
- Status ConnectToCluster(KuduClient* client, const MonoTime& deadline);
+ Status ConnectToCluster(
+ KuduClient* client,
+ const MonoTime& deadline,
+ rpc::CredentialsPolicy creds_policy = rpc::CredentialsPolicy::ANY_CREDENTIALS);
std::shared_ptr<master::MasterServiceProxy> master_proxy() const;
@@ -230,11 +234,13 @@ class KuduClient::Data {
// Ref-counted RPC instance: since 'ConnectToClusterAsync' call
// is asynchronous, we need to hold a reference in this class
// itself, as to avoid a "use-after-free" scenario.
- scoped_refptr<internal::ConnectToClusterRpc> leader_master_rpc_;
- std::vector<StatusCallback> leader_master_callbacks_;
+ scoped_refptr<internal::ConnectToClusterRpc> leader_master_rpc_any_creds_;
+ std::vector<StatusCallback> leader_master_callbacks_any_creds_;
+ scoped_refptr<internal::ConnectToClusterRpc> leader_master_rpc_primary_creds_;
+ std::vector<StatusCallback> leader_master_callbacks_primary_creds_;
- // Protects 'leader_master_rpc_', 'leader_master_hostport_',
- // and master_proxy_
+ // Protects 'leader_master_rpc_{any,primary}_creds_',
+ // 'leader_master_hostport_', and 'master_proxy_'.
//
// See: KuduClient::Data::ConnectToClusterAsync for a more
// in-depth explanation of why this is needed and how it works.
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/client/master_rpc.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/master_rpc.cc b/src/kudu/client/master_rpc.cc
index 61e45c5..e78c349 100644
--- a/src/kudu/client/master_rpc.cc
+++ b/src/kudu/client/master_rpc.cc
@@ -19,15 +19,17 @@
#include "kudu/client/master_rpc.h"
-#include <boost/bind.hpp>
#include <mutex>
+#include <boost/bind.hpp>
+
#include "kudu/common/wire_protocol.h"
#include "kudu/common/wire_protocol.pb.h"
#include "kudu/gutil/bind.h"
#include "kudu/gutil/strings/join.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/master/master.proxy.h"
+#include "kudu/rpc/rpc_controller.h"
#include "kudu/util/net/net_util.h"
#include "kudu/util/scoped_cleanup.h"
@@ -43,6 +45,7 @@ using kudu::master::GetMasterRegistrationRequestPB;
using kudu::master::GetMasterRegistrationResponsePB;
using kudu::master::MasterErrorPB;
using kudu::master::MasterServiceProxy;
+using kudu::rpc::CredentialsPolicy;
using kudu::rpc::Messenger;
using kudu::rpc::Rpc;
@@ -68,6 +71,7 @@ class ConnectToMasterRpc : public rpc::Rpc {
const Sockaddr& addr,
const MonoTime& deadline,
std::shared_ptr<rpc::Messenger> messenger,
+ CredentialsPolicy creds_policy,
ConnectToMasterResponsePB* out);
~ConnectToMasterRpc();
@@ -94,33 +98,37 @@ class ConnectToMasterRpc : public rpc::Rpc {
};
-ConnectToMasterRpc::ConnectToMasterRpc(
- StatusCallback user_cb, const Sockaddr& addr,const MonoTime& deadline,
- shared_ptr<Messenger> messenger, ConnectToMasterResponsePB* out)
- : Rpc(deadline, std::move(messenger)),
- user_cb_(std::move(user_cb)),
- addr_(addr),
- out_(DCHECK_NOTNULL(out)) {}
+ConnectToMasterRpc::ConnectToMasterRpc(StatusCallback user_cb,
+ const Sockaddr& addr,
+ const MonoTime& deadline,
+ shared_ptr<Messenger> messenger,
+ rpc::CredentialsPolicy creds_policy,
+ ConnectToMasterResponsePB* out)
+ : Rpc(deadline, std::move(messenger)),
+ user_cb_(std::move(user_cb)),
+ addr_(addr),
+ out_(DCHECK_NOTNULL(out)) {
+ mutable_retrier()->mutable_controller()->set_credentials_policy(creds_policy);
+}
ConnectToMasterRpc::~ConnectToMasterRpc() {
}
void ConnectToMasterRpc::SendRpc() {
- MasterServiceProxy proxy(retrier().messenger(),
- addr_);
- rpc::RpcController* rpc = mutable_retrier()->mutable_controller();
+ MasterServiceProxy proxy(retrier().messenger(), addr_);
+ rpc::RpcController* controller = mutable_retrier()->mutable_controller();
// TODO(todd): should this be setting an RPC call deadline based on 'deadline'?
// it doesn't seem to be.
if (!trying_old_rpc_) {
ConnectToMasterRequestPB req;
- rpc->RequireServerFeature(master::MasterFeatures::CONNECT_TO_MASTER);
- proxy.ConnectToMasterAsync(req, out_, rpc,
+ controller->RequireServerFeature(master::MasterFeatures::CONNECT_TO_MASTER);
+ proxy.ConnectToMasterAsync(req, out_, controller,
boost::bind(&ConnectToMasterRpc::SendRpcCb,
this,
Status::OK()));
} else {
GetMasterRegistrationRequestPB req;
- proxy.GetMasterRegistrationAsync(req, &old_rpc_resp_, rpc,
+ proxy.GetMasterRegistrationAsync(req, &old_rpc_resp_, controller,
boost::bind(&ConnectToMasterRpc::SendRpcCb,
this,
Status::OK()));
@@ -192,7 +200,8 @@ ConnectToClusterRpc::ConnectToClusterRpc(LeaderCallback user_cb,
vector<Sockaddr> addrs,
MonoTime deadline,
MonoDelta rpc_timeout,
- shared_ptr<Messenger> messenger)
+ shared_ptr<Messenger> messenger,
+ rpc::CredentialsPolicy creds_policy)
: Rpc(deadline, std::move(messenger)),
user_cb_(std::move(user_cb)),
addrs_(std::move(addrs)),
@@ -203,6 +212,7 @@ ConnectToClusterRpc::ConnectToClusterRpc(LeaderCallback user_cb,
// Using resize instead of reserve to explicitly initialized the values.
responses_.resize(addrs_.size());
+ mutable_retrier()->mutable_controller()->set_credentials_policy(creds_policy);
}
ConnectToClusterRpc::~ConnectToClusterRpc() {
@@ -231,6 +241,7 @@ void ConnectToClusterRpc::SendRpc() {
addrs_[i],
actual_deadline,
retrier().messenger(),
+ retrier().controller().credentials_policy(),
&responses_[i]);
rpc->SendRpc();
++pending_responses_;
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/client/master_rpc.h
----------------------------------------------------------------------
diff --git a/src/kudu/client/master_rpc.h b/src/kudu/client/master_rpc.h
index 1e55bea..b57d137 100644
--- a/src/kudu/client/master_rpc.h
+++ b/src/kudu/client/master_rpc.h
@@ -25,6 +25,7 @@
#include "kudu/gutil/ref_counted.h"
#include "kudu/master/master.pb.h"
#include "kudu/rpc/rpc.h"
+#include "kudu/rpc/rpc_controller.h"
#include "kudu/util/locks.h"
#include "kudu/util/net/net_util.h"
#include "kudu/util/net/sockaddr.h"
@@ -79,7 +80,9 @@ class ConnectToClusterRpc : public rpc::Rpc,
std::vector<Sockaddr> addrs,
MonoTime deadline,
MonoDelta rpc_timeout,
- std::shared_ptr<rpc::Messenger> messenger);
+ std::shared_ptr<rpc::Messenger> messenger,
+ rpc::CredentialsPolicy creds_policy =
+ rpc::CredentialsPolicy::ANY_CREDENTIALS);
virtual void SendRpc() OVERRIDE;
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/client/meta_cache.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/meta_cache.cc b/src/kudu/client/meta_cache.cc
index 0b8b896..b58f45c 100644
--- a/src/kudu/client/meta_cache.cc
+++ b/src/kudu/client/meta_cache.cc
@@ -17,9 +17,14 @@
#include "kudu/client/meta_cache.h"
+#include <map>
+#include <mutex>
+#include <set>
+#include <string>
+#include <vector>
+
#include <boost/bind.hpp>
#include <glog/logging.h>
-#include <mutex>
#include "kudu/client/client.h"
#include "kudu/client/client-internal.h"
@@ -32,6 +37,8 @@
#include "kudu/master/master.proxy.h"
#include "kudu/rpc/messenger.h"
#include "kudu/rpc/rpc.h"
+#include "kudu/rpc/rpc_controller.h"
+#include "kudu/rpc/rpc_header.pb.h"
#include "kudu/tserver/tserver_service.proxy.h"
#include "kudu/util/logging.h"
#include "kudu/util/net/dns_resolver.h"
@@ -53,8 +60,10 @@ using master::MasterServiceProxy;
using master::TabletLocationsPB;
using master::TabletLocationsPB_ReplicaPB;
using master::TSInfoPB;
+using rpc::CredentialsPolicy;
using rpc::Messenger;
using rpc::Rpc;
+using rpc::ErrorStatusPB;
using tserver::TabletServerServiceProxy;
namespace client {
@@ -537,7 +546,7 @@ class LookupRpc : public Rpc {
return table_->client()->data_->master_proxy();
}
- void ResetMasterLeaderAndRetry();
+ void ResetMasterLeaderAndRetry(CredentialsPolicy creds_policy);
void NewLeaderMasterDeterminedCb(const Status& status);
@@ -671,12 +680,12 @@ string LookupRpc::ToString() const {
num_attempts());
}
-void LookupRpc::ResetMasterLeaderAndRetry() {
+void LookupRpc::ResetMasterLeaderAndRetry(CredentialsPolicy creds_policy) {
table_->client()->data_->ConnectToClusterAsync(
table_->client(),
retrier().deadline(),
- Bind(&LookupRpc::NewLeaderMasterDeterminedCb,
- Unretained(this)));
+ Bind(&LookupRpc::NewLeaderMasterDeterminedCb, Unretained(this)),
+ creds_policy);
}
void LookupRpc::NewLeaderMasterDeterminedCb(const Status& status) {
@@ -710,7 +719,7 @@ void LookupRpc::SendRpcCb(const Status& status) {
resp_.error().code() == master::MasterErrorPB::CATALOG_MANAGER_NOT_INITIALIZED) {
if (meta_cache_->client_->IsMultiMaster()) {
KLOG_EVERY_N_SECS(WARNING, 1) << "Leader Master has changed, re-trying...";
- ResetMasterLeaderAndRetry();
+ ResetMasterLeaderAndRetry(CredentialsPolicy::ANY_CREDENTIALS);
ignore_result(delete_me.release());
return;
}
@@ -723,7 +732,7 @@ void LookupRpc::SendRpcCb(const Status& status) {
if (MonoTime::Now() < retrier().deadline()) {
if (meta_cache_->client_->IsMultiMaster()) {
KLOG_EVERY_N_SECS(WARNING, 1) << "Leader Master timed out, re-trying...";
- ResetMasterLeaderAndRetry();
+ ResetMasterLeaderAndRetry(CredentialsPolicy::ANY_CREDENTIALS);
ignore_result(delete_me.release());
return;
}
@@ -738,7 +747,20 @@ void LookupRpc::SendRpcCb(const Status& status) {
if (meta_cache_->client_->IsMultiMaster()) {
KLOG_EVERY_N_SECS(WARNING, 1) << "Encountered a network error from the Master: "
<< new_status.ToString() << ", retrying...";
- ResetMasterLeaderAndRetry();
+ ResetMasterLeaderAndRetry(CredentialsPolicy::ANY_CREDENTIALS);
+ ignore_result(delete_me.release());
+ return;
+ }
+ }
+
+ if (new_status.IsNotAuthorized()) {
+ const rpc::RpcController& controller(retrier().controller());
+ const ErrorStatusPB* err = controller.error_response();
+ if (err && err->has_code() &&
+ err->code() == ErrorStatusPB::FATAL_INVALID_AUTHENTICATION_TOKEN) {
+ KLOG_EVERY_N_SECS(INFO, 1)
+ << "Re-connecting to cluster in attempt to get new authn token";
+ ResetMasterLeaderAndRetry(CredentialsPolicy::PRIMARY_CREDENTIALS);
ignore_result(delete_me.release());
return;
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/client/scanner-internal.cc
----------------------------------------------------------------------
diff --git a/src/kudu/client/scanner-internal.cc b/src/kudu/client/scanner-internal.cc
index 21bf8ed..bcaf994 100644
--- a/src/kudu/client/scanner-internal.cc
+++ b/src/kudu/client/scanner-internal.cc
@@ -30,6 +30,7 @@
#include "kudu/common/common.pb.h"
#include "kudu/common/wire_protocol.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/rpc/messenger.h"
#include "kudu/rpc/rpc_controller.h"
#include "kudu/util/hexdump.h"
@@ -41,9 +42,9 @@ using std::string;
namespace kudu {
+using rpc::CredentialsPolicy;
using rpc::RpcController;
using strings::Substitute;
-using strings::SubstituteAndAppend;
using tserver::NewScanRequestPB;
using tserver::TabletServerFeatures;
@@ -82,6 +83,7 @@ Status KuduScanner::Data::HandleError(const ScanRpcStatus& err,
bool mark_locations_stale = false;
bool can_retry = true;
bool backoff = false;
+ bool reacquire_authn_token = false;
switch (err.result) {
case ScanRpcStatus::SERVICE_UNAVAILABLE:
backoff = true;
@@ -93,6 +95,11 @@ Status KuduScanner::Data::HandleError(const ScanRpcStatus& err,
break;
case ScanRpcStatus::SCANNER_EXPIRED:
break;
+ case ScanRpcStatus::RPC_INVALID_AUTHENTICATION_TOKEN:
+ // Usually this happens if doing an RPC call with an expired auth token.
+ // Retrying with a new authn token should help.
+ reacquire_authn_token = true;
+ break;
case ScanRpcStatus::TABLET_NOT_RUNNING:
blacklist_location = true;
break;
@@ -121,6 +128,18 @@ Status KuduScanner::Data::HandleError(const ScanRpcStatus& err,
remote_->MarkStale();
}
+ if (reacquire_authn_token) {
+ // Re-connect to the cluster to get a new authn token.
+ KuduClient* c = table_->client();
+ const Status& s = c->data_->ConnectToCluster(
+ c, deadline, CredentialsPolicy::PRIMARY_CREDENTIALS);
+ if (!s.ok()) {
+ KLOG_EVERY_N_SECS(WARNING, 1)
+ << "Couldn't reconnect to the cluster: " << s.ToString();
+ backoff = true;
+ }
+ }
+
if (backoff) {
MonoDelta sleep =
KuduClient::Data::ComputeExponentialBackoff(scan_attempts_);
@@ -136,6 +155,7 @@ Status KuduScanner::Data::HandleError(const ScanRpcStatus& err,
<< sleep.ToString() << "; attempt " << scan_attempts_;
SleepFor(sleep);
}
+
if (can_retry) {
return Status::OK();
}
@@ -182,6 +202,16 @@ ScanRpcStatus KuduScanner::Data::AnalyzeResponse(const Status& rpc_status,
}
}
+ if (rpc_status.IsNotAuthorized() && controller_.error_response()) {
+ switch (controller_.error_response()->code()) {
+ case rpc::ErrorStatusPB::FATAL_INVALID_AUTHENTICATION_TOKEN:
+ return ScanRpcStatus{
+ ScanRpcStatus::RPC_INVALID_AUTHENTICATION_TOKEN, rpc_status};
+ default:
+ return ScanRpcStatus{ScanRpcStatus::OTHER_TS_ERROR, rpc_status};
+ }
+ }
+
if (rpc_status.IsTimedOut()) {
if (overall_deadline == deadline) {
return ScanRpcStatus{ScanRpcStatus::OVERALL_DEADLINE_EXCEEDED, rpc_status};
@@ -340,9 +370,8 @@ Status KuduScanner::Data::OpenTablet(const string& partition_key,
// No more tablets in the table.
partition_pruner_.RemovePartitionKeyRange("");
return Status::OK();
- } else {
- RETURN_NOT_OK(s);
}
+ RETURN_NOT_OK(s);
// Check if the meta cache returned a tablet covering a partition key range past
// what we asked for. This can happen if the requested partition key falls
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/client/scanner-internal.h
----------------------------------------------------------------------
diff --git a/src/kudu/client/scanner-internal.h b/src/kudu/client/scanner-internal.h
index e3f3418..4ca99f2 100644
--- a/src/kudu/client/scanner-internal.h
+++ b/src/kudu/client/scanner-internal.h
@@ -60,6 +60,10 @@ struct ScanRpcStatus {
// on other hosts.
RPC_DEADLINE_EXCEEDED,
+ // The authentication token supplied by the client is invalid. Most likely,
+ // the token has expired.
+ RPC_INVALID_AUTHENTICATION_TOKEN,
+
// Another RPC-system error (e.g. NetworkError because the TS was down).
RPC_ERROR,
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/integration-tests/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/CMakeLists.txt b/src/kudu/integration-tests/CMakeLists.txt
index 5cb1500..ada08e2 100644
--- a/src/kudu/integration-tests/CMakeLists.txt
+++ b/src/kudu/integration-tests/CMakeLists.txt
@@ -53,6 +53,7 @@ set(KUDU_TEST_LINK_LIBS integration-tests ${KUDU_MIN_TEST_LIBS})
ADD_KUDU_TEST(all_types-itest RESOURCE_LOCK "master-rpc-ports")
ADD_KUDU_TEST(alter_table-randomized-test)
ADD_KUDU_TEST(alter_table-test)
+ADD_KUDU_TEST(authn_token_expire-itest)
ADD_KUDU_TEST(catalog_manager_tsk-itest)
ADD_KUDU_TEST(client_failover-itest)
ADD_KUDU_TEST(client-stress-test
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/integration-tests/authn_token_expire-itest.cc
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/authn_token_expire-itest.cc b/src/kudu/integration-tests/authn_token_expire-itest.cc
new file mode 100644
index 0000000..d82cf12
--- /dev/null
+++ b/src/kudu/integration-tests/authn_token_expire-itest.cc
@@ -0,0 +1,320 @@
+// 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 <memory>
+#include <string>
+
+#include <gflags/gflags_declare.h>
+#include <glog/logging.h>
+#include <gtest/gtest.h>
+
+#include "kudu/client/client.h"
+#include "kudu/client/client-test-util.h"
+#include "kudu/client/shared_ptr.h"
+#include "kudu/common/partial_row.h"
+#include "kudu/gutil/stringprintf.h"
+#include "kudu/gutil/strings/substitute.h"
+#include "kudu/master/catalog_manager.h"
+#include "kudu/master/master.h"
+#include "kudu/master/master.pb.h"
+#include "kudu/master/mini_master.h"
+#include "kudu/integration-tests/cluster_itest_util.h"
+#include "kudu/integration-tests/cluster_verifier.h"
+#include "kudu/integration-tests/external_mini_cluster.h"
+#include "kudu/integration-tests/test_workload.h"
+#include "kudu/server/hybrid_clock.h"
+#include "kudu/tablet/key_value_test_schema.h"
+#include "kudu/tablet/tablet.h"
+#include "kudu/tablet/tablet_replica.h"
+#include "kudu/tserver/mini_tablet_server.h"
+#include "kudu/tserver/tablet_server.h"
+#include "kudu/tserver/ts_tablet_manager.h"
+#include "kudu/util/monotime.h"
+#include "kudu/util/test_util.h"
+
+DECLARE_bool(rpc_reopen_outbound_connections);
+
+using std::string;
+using std::unique_ptr;
+using strings::Substitute;
+
+namespace kudu {
+namespace client {
+
+using sp::shared_ptr;
+
+namespace {
+
+// Create a table with the specified name (no replication) and specified
+// number of hash partitions.
+Status CreateTable(KuduClient* client,
+ const string& table_name,
+ const KuduSchema& schema,
+ int num_hash_partitions) {
+ unique_ptr<KuduTableCreator> table_creator(client->NewTableCreator());
+ return table_creator->table_name(table_name)
+ .schema(&schema)
+ .add_hash_partitions({ "key" }, num_hash_partitions)
+ .num_replicas(1)
+ .Create();
+}
+
+unique_ptr<KuduInsert> BuildTestRow(KuduTable* table, int index) {
+ unique_ptr<KuduInsert> insert(table->NewInsert());
+ KuduPartialRow* row = insert->mutable_row();
+ CHECK_OK(row->SetInt32(0, index));
+ CHECK_OK(row->SetInt32(1, index * 2));
+ return insert;
+}
+
+// Insert given number of tests rows into the default test table in the context
+// of a new session.
+Status InsertTestRows(KuduClient* client, KuduTable* table,
+ int num_rows, int first_row = 0) {
+ shared_ptr<KuduSession> session = client->NewSession();
+ RETURN_NOT_OK(session->SetFlushMode(KuduSession::AUTO_FLUSH_SYNC));
+ session->SetTimeoutMillis(60000);
+ for (int i = first_row; i < num_rows + first_row; ++i) {
+ unique_ptr<KuduInsert> insert(BuildTestRow(table, i));
+ RETURN_NOT_OK(session->Apply(insert.release()));
+ }
+ return session->Flush();
+}
+
+} // anonymous namespace
+
+class AuthnTokenExpireITestBase : public KuduTest {
+ public:
+ AuthnTokenExpireITestBase(int num_tablet_servers,
+ int64_t token_validity_seconds)
+ : num_tablet_servers_(num_tablet_servers),
+ token_validity_seconds_(token_validity_seconds),
+ schema_(client::KuduSchemaFromSchema(CreateKeyValueTestSchema())) {
+ cluster_opts_.num_tablet_servers = num_tablet_servers_;
+ cluster_opts_.enable_kerberos = true;
+ }
+
+ void SetUp() override {
+ KuduTest::SetUp();
+ cluster_.reset(new ExternalMiniCluster(cluster_opts_));
+ ASSERT_OK(cluster_->Start());
+ }
+
+ void TearDown() override {
+ if (cluster_) {
+ cluster_->Shutdown();
+ }
+ KuduTest::TearDown();
+ }
+
+ protected:
+ const int num_tablet_servers_;
+ const int64_t token_validity_seconds_;
+ KuduSchema schema_;
+ ExternalMiniClusterOptions cluster_opts_;
+ shared_ptr<ExternalMiniCluster> cluster_;
+};
+
+
+class AuthnTokenExpireITest : public AuthnTokenExpireITestBase {
+ public:
+ AuthnTokenExpireITest()
+ : AuthnTokenExpireITestBase(3, 2) {
+ // Masters and tservers inject FATAL_INVALID_AUTHENTICATION_TOKEN errors.
+ // The client should retry the operation again and eventually it should
+ // succeed even with the high ratio of injected errors.
+ cluster_opts_.extra_master_flags = {
+ "--rpc_inject_invalid_authn_token_ratio=0.5",
+
+ // In addition to very short authn token TTL, rotate token signing keys as
+ // often as possible: we want to cover TSK propagation-related scenarios
+ // as well (i.e. possible ERROR_UNAVAILABLE errors from tservers) upon
+ // a new authn token re-acquisitions and retried RPCs.
+ "--tsk_rotation_seconds=1",
+ Substitute("--authn_token_validity_seconds=$0", token_validity_seconds_),
+ };
+
+ cluster_opts_.extra_tserver_flags = {
+ "--rpc_inject_invalid_authn_token_ratio=0.5",
+
+ // Decreasing TS->master heartbeat interval speeds up the test.
+ "--heartbeat_interval_ms=10",
+ };
+ }
+
+ void SetUp() override {
+ AuthnTokenExpireITestBase::SetUp();
+ }
+};
+
+
+// Make sure authn token is re-acquired on certain scenarios upon restarting
+// tablet servers.
+TEST_F(AuthnTokenExpireITest, RestartTabletServers) {
+ const string table_name = "authn-token-expire-restart-tablet-servers";
+
+ // Create and open one table, keeping it open over the component restarts.
+ shared_ptr<KuduClient> client;
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client));
+ ASSERT_OK(CreateTable(client.get(), table_name, schema_, num_tablet_servers_));
+ shared_ptr<KuduTable> table;
+ ASSERT_OK(client->OpenTable(table_name, &table));
+ ASSERT_OK(InsertTestRows(client.get(), table.get(),
+ num_tablet_servers_, num_tablet_servers_ * 0));
+
+ // Restart all tablet servers.
+ for (int i = 0; i < cluster_->num_tablet_servers(); ++i) {
+ auto server = cluster_->tablet_server(i);
+ ASSERT_NE(nullptr, server);
+ server->Shutdown();
+ ASSERT_OK(server->Restart());
+ }
+ SleepFor(MonoDelta::FromSeconds(token_validity_seconds_ + 1));
+
+ ASSERT_OK(InsertTestRows(client.get(), table.get(),
+ num_tablet_servers_, num_tablet_servers_ * 1));
+ SleepFor(MonoDelta::FromSeconds(token_validity_seconds_ + 1));
+ // Make sure to insert a row into all tablets to make an RPC call to every
+ // tablet server hosting the table.
+ ASSERT_OK(InsertTestRows(client.get(), table.get(),
+ num_tablet_servers_, num_tablet_servers_ * 2));
+}
+
+// Make sure authn token is re-acquired on certain scenarios upon restarting
+// both masters and tablet servers.
+TEST_F(AuthnTokenExpireITest, RestartCluster) {
+ const string table_name = "authn-token-expire-restart-cluster";
+
+ shared_ptr<KuduClient> client;
+ ASSERT_OK(cluster_->CreateClient(nullptr, &client));
+ ASSERT_OK(CreateTable(client.get(), table_name, schema_, num_tablet_servers_));
+ shared_ptr<KuduTable> table;
+ ASSERT_OK(client->OpenTable(table_name, &table));
+ ASSERT_OK(InsertTestRows(client.get(), table.get(),
+ num_tablet_servers_, num_tablet_servers_ * 0));
+
+ // Restart all Kudu server-side components: masters and tablet servers.
+ cluster_->Shutdown();
+ ASSERT_OK(cluster_->Restart());
+ SleepFor(MonoDelta::FromSeconds(token_validity_seconds_ + 1));
+
+ ASSERT_OK(InsertTestRows(client.get(), table.get(),
+ num_tablet_servers_, num_tablet_servers_ * 1));
+ SleepFor(MonoDelta::FromSeconds(token_validity_seconds_ + 1));
+ // Make sure to insert a row into all tablets to make an RPC call to every
+ // tablet server hosting the table.
+ ASSERT_OK(InsertTestRows(client.get(), table.get(),
+ num_tablet_servers_, num_tablet_servers_ * 2));
+}
+
+// Run the workload and check that client retries INVALID_AUTH_TOKEN errors,
+// eventually succeeding with its RPCs.
+TEST_F(AuthnTokenExpireITest, InvalidTokenDuringWorkload) {
+ static const int32_t kTimeoutMs = 10 * 60 * 1000;
+
+ if (!AllowSlowTests()) {
+ LOG(WARNING) << "test is skipped; set KUDU_ALLOW_SLOW_TESTS=1 to run";
+ return;
+ }
+
+ // Close an already established idle connection to the server and open
+ // a new one upon making another call to the same server. This is to force
+ // authn token verification at every RPC.
+ FLAGS_rpc_reopen_outbound_connections = true;
+
+ TestWorkload w(cluster_.get());
+ w.set_client_default_admin_operation_timeout_millis(kTimeoutMs);
+ w.set_client_default_rpc_timeout_millis(kTimeoutMs);
+ w.set_num_replicas(num_tablet_servers_);
+ w.set_num_read_threads(3);
+ w.set_read_timeout_millis(kTimeoutMs);
+ w.set_num_write_threads(3);
+ w.set_write_batch_size(64);
+ w.set_write_timeout_millis(kTimeoutMs);
+ w.set_write_pattern(TestWorkload::INSERT_SEQUENTIAL_ROWS);
+
+ w.Setup();
+ w.Start();
+ SleepFor(MonoDelta::FromSeconds(8 * token_validity_seconds_));
+ w.StopAndJoin();
+
+ ClusterVerifier v(cluster_.get());
+ v.SetOperationsTimeout(MonoDelta::FromSeconds(5 * 60));
+ NO_FATALS(v.CheckRowCount(
+ w.table_name(), ClusterVerifier::EXACTLY, w.rows_inserted()));
+ ASSERT_OK(w.Cleanup());
+
+ NO_FATALS(cluster_->AssertNoCrashes());
+}
+
+class AuthnTokenReacquireITest : public AuthnTokenExpireITestBase {
+ public:
+ AuthnTokenReacquireITest()
+ : AuthnTokenExpireITestBase(3, 2) {
+
+ cluster_opts_.extra_master_flags = {
+ Substitute("--authn_token_validity_seconds=$0", token_validity_seconds_),
+ };
+
+ cluster_opts_.extra_tserver_flags = {
+ // Decreasing TS->master heartbeat interval speeds up the test.
+ "--heartbeat_interval_ms=10",
+ };
+ }
+};
+
+// This test verifies that the token re-acquire logic behaves correctly in case
+// if the connection to the master is established using previously acquired
+// authn token. The master has particular constraint to prohibit re-issuing
+// a new authn token over a connection established with authn token itself
+// (otherwise, a authn token would never effectively expire).
+TEST_F(AuthnTokenReacquireITest, ConnectionToMasterViaAuthnToken) {
+ const string table_name = "authn-token-reacquire";
+
+ // Create a client and perform some basic operations to acquire authn token.
+ shared_ptr<KuduClient> client;
+ ASSERT_OK(cluster_->CreateClient(
+ &KuduClientBuilder()
+ .default_admin_operation_timeout(MonoDelta::FromSeconds(60))
+ .default_rpc_timeout(MonoDelta::FromSeconds(60)),
+ &client));
+ ASSERT_OK(CreateTable(client.get(), table_name, schema_, num_tablet_servers_));
+
+ // Restart the master and the tablet servers to make sure all connectons
+ // between the client and the servers are closed.
+ cluster_->Shutdown();
+ ASSERT_OK(cluster_->Restart());
+
+ // Perform some operations using already existing token. The crux here is to
+ // establish a connection to the master server where client is authenticated
+ // via the authn token, not Kerberos credentials.
+ shared_ptr<KuduTable> table;
+ ASSERT_OK(client->OpenTable(table_name, &table));
+
+ // Let the authn token to expire.
+ SleepFor(MonoDelta::FromSeconds(token_validity_seconds_ + 1));
+
+ // Here a new authn token should be automatically acquired upon receiving
+ // FATAL_INVALID_AUTHENTICATION_TOKEN error. To get a new token it's necessary
+ // to establish a new connection to the master _not_ using the authn token
+ // as client-side credentials.
+ ASSERT_OK(InsertTestRows(client.get(), table.get(), num_tablet_servers_));
+ NO_FATALS(cluster_->AssertNoCrashes());
+}
+
+} // namespace client
+} // namespace kudu
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/integration-tests/cluster_verifier.h
----------------------------------------------------------------------
diff --git a/src/kudu/integration-tests/cluster_verifier.h b/src/kudu/integration-tests/cluster_verifier.h
index 14d5499..50aa8b0 100644
--- a/src/kudu/integration-tests/cluster_verifier.h
+++ b/src/kudu/integration-tests/cluster_verifier.h
@@ -90,10 +90,10 @@ class ClusterVerifier {
ExternalMiniCluster* cluster_;
- MonoDelta operations_timeout_;
-
ChecksumOptions checksum_options_;
+ MonoDelta operations_timeout_;
+
DISALLOW_COPY_AND_ASSIGN(ClusterVerifier);
};
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/rpc/client_negotiation.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/client_negotiation.cc b/src/kudu/rpc/client_negotiation.cc
index 5d5c0f0..8ce5190 100644
--- a/src/kudu/rpc/client_negotiation.cc
+++ b/src/kudu/rpc/client_negotiation.cc
@@ -94,9 +94,10 @@ static Status StatusFromRpcError(const ErrorStatusPB& error) {
if (PREDICT_FALSE(!error.has_code())) {
return Status::RuntimeError(error.message());
}
- string code_name = ErrorStatusPB::RpcErrorCodePB_Name(error.code());
+ const string code_name = ErrorStatusPB::RpcErrorCodePB_Name(error.code());
switch (error.code()) {
- case ErrorStatusPB_RpcErrorCodePB_FATAL_UNAUTHORIZED:
+ case ErrorStatusPB_RpcErrorCodePB_FATAL_UNAUTHORIZED: // fall-through
+ case ErrorStatusPB_RpcErrorCodePB_FATAL_INVALID_AUTHENTICATION_TOKEN:
return Status::NotAuthorized(code_name, error.message());
case ErrorStatusPB_RpcErrorCodePB_ERROR_UNAVAILABLE:
return Status::ServiceUnavailable(code_name, error.message());
@@ -153,7 +154,7 @@ void ClientNegotiation::set_deadline(const MonoTime& deadline) {
deadline_ = deadline;
}
-Status ClientNegotiation::Negotiate() {
+Status ClientNegotiation::Negotiate(unique_ptr<ErrorStatusPB>* rpc_error) {
TRACE("Beginning negotiation");
// Ensure we can use blocking calls on the socket during negotiation.
@@ -167,7 +168,7 @@ Status ClientNegotiation::Negotiate() {
{ // Step 2: send and receive the NEGOTIATE step messages.
RETURN_NOT_OK(SendNegotiate());
NegotiatePB response;
- RETURN_NOT_OK(RecvNegotiatePB(&response, &recv_buf));
+ RETURN_NOT_OK(RecvNegotiatePB(&response, &recv_buf, rpc_error));
RETURN_NOT_OK(HandleNegotiate(response));
TRACE("Negotiated authn=$0", AuthenticationTypeToString(negotiated_authn_));
}
@@ -195,7 +196,7 @@ Status ClientNegotiation::Negotiate() {
while (s.IsIncomplete()) {
NegotiatePB response;
- RETURN_NOT_OK(RecvNegotiatePB(&response, &recv_buf));
+ RETURN_NOT_OK(RecvNegotiatePB(&response, &recv_buf, rpc_error));
s = HandleTlsHandshake(response);
}
RETURN_NOT_OK(s);
@@ -205,10 +206,10 @@ Status ClientNegotiation::Negotiate() {
// Step 4: Authentication
switch (negotiated_authn_) {
case AuthenticationType::SASL:
- RETURN_NOT_OK(AuthenticateBySasl(&recv_buf));
+ RETURN_NOT_OK(AuthenticateBySasl(&recv_buf, rpc_error));
break;
case AuthenticationType::TOKEN:
- RETURN_NOT_OK(AuthenticateByToken(&recv_buf));
+ RETURN_NOT_OK(AuthenticateByToken(&recv_buf, rpc_error));
break;
case AuthenticationType::CERTIFICATE:
// The TLS handshake has already authenticated the server.
@@ -235,14 +236,16 @@ Status ClientNegotiation::SendNegotiatePB(const NegotiatePB& msg) {
return SendFramedMessageBlocking(socket(), header, msg, deadline_);
}
-Status ClientNegotiation::RecvNegotiatePB(NegotiatePB* msg, faststring* buffer) {
+Status ClientNegotiation::RecvNegotiatePB(NegotiatePB* msg,
+ faststring* buffer,
+ unique_ptr<ErrorStatusPB>* rpc_error) {
ResponseHeader header;
Slice param_buf;
RETURN_NOT_OK(ReceiveFramedMessageBlocking(socket(), buffer, &header, ¶m_buf, deadline_));
RETURN_NOT_OK(helper_.CheckNegotiateCallId(header.call_id()));
if (header.is_error()) {
- return ParseError(param_buf);
+ return ParseError(param_buf, rpc_error);
}
RETURN_NOT_OK(helper_.ParseNegotiatePB(param_buf, msg));
@@ -250,14 +253,19 @@ Status ClientNegotiation::RecvNegotiatePB(NegotiatePB* msg, faststring* buffer)
return Status::OK();
}
-Status ClientNegotiation::ParseError(const Slice& err_data) {
- ErrorStatusPB error;
- if (!error.ParseFromArray(err_data.data(), err_data.size())) {
+Status ClientNegotiation::ParseError(const Slice& err_data,
+ unique_ptr<ErrorStatusPB>* rpc_error) {
+ unique_ptr<ErrorStatusPB> error(new ErrorStatusPB);
+ if (!error->ParseFromArray(err_data.data(), err_data.size())) {
return Status::IOError("invalid error response, missing fields",
- error.InitializationErrorString());
+ error->InitializationErrorString());
}
- Status s = StatusFromRpcError(error);
+ Status s = StatusFromRpcError(*error);
TRACE("Received error response from server: $0", s.ToString());
+
+ if (rpc_error) {
+ rpc_error->swap(error);
+ }
return s;
}
@@ -487,7 +495,8 @@ Status ClientNegotiation::HandleTlsHandshake(const NegotiatePB& response) {
return tls_handshake_.Finish(&socket_);
}
-Status ClientNegotiation::AuthenticateBySasl(faststring* recv_buf) {
+Status ClientNegotiation::AuthenticateBySasl(faststring* recv_buf,
+ unique_ptr<ErrorStatusPB>* rpc_error) {
RETURN_NOT_OK(InitSaslClient());
Status s = SendSaslInitiate();
@@ -495,7 +504,7 @@ Status ClientNegotiation::AuthenticateBySasl(faststring* recv_buf) {
// challenge step is required, or OK if a SASL_SUCCESS message is expected.
while (s.IsIncomplete()) {
NegotiatePB challenge;
- RETURN_NOT_OK(RecvNegotiatePB(&challenge, recv_buf));
+ RETURN_NOT_OK(RecvNegotiatePB(&challenge, recv_buf, rpc_error));
s = HandleSaslChallenge(challenge);
}
@@ -504,11 +513,12 @@ Status ClientNegotiation::AuthenticateBySasl(faststring* recv_buf) {
// Server challenges are over; we now expect the success message.
NegotiatePB success;
- RETURN_NOT_OK(RecvNegotiatePB(&success, recv_buf));
+ RETURN_NOT_OK(RecvNegotiatePB(&success, recv_buf, rpc_error));
return HandleSaslSuccess(success);
}
-Status ClientNegotiation::AuthenticateByToken(faststring* recv_buf) {
+Status ClientNegotiation::AuthenticateByToken(faststring* recv_buf,
+ unique_ptr<ErrorStatusPB>* rpc_error) {
// Sanity check that TLS has been negotiated. Sending the token on an
// unencrypted channel is a big no-no.
CHECK(tls_negotiated_);
@@ -521,7 +531,7 @@ Status ClientNegotiation::AuthenticateByToken(faststring* recv_buf) {
pb.Clear();
// Check that the server responds with a non-error TOKEN_EXCHANGE message.
- RETURN_NOT_OK(RecvNegotiatePB(&pb, recv_buf));
+ RETURN_NOT_OK(RecvNegotiatePB(&pb, recv_buf, rpc_error));
if (pb.step() != NegotiatePB::TOKEN_EXCHANGE) {
return Status::NotAuthorized("expected TOKEN_EXCHANGE step",
NegotiatePB::NegotiateStep_Name(pb.step()));
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/rpc/client_negotiation.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/client_negotiation.h b/src/kudu/rpc/client_negotiation.h
index 701076b..aa8c6d9 100644
--- a/src/kudu/rpc/client_negotiation.h
+++ b/src/kudu/rpc/client_negotiation.h
@@ -122,7 +122,7 @@ class ClientNegotiation {
//
// Returns OK on success, otherwise may return NotAuthorized, NotSupported, or
// another non-OK status.
- Status Negotiate();
+ Status Negotiate(std::unique_ptr<ErrorStatusPB>* rpc_error = nullptr);
// SASL callback for plugin options, supported mechanisms, etc.
// Returns SASL_FAIL if the option is not handled, which does not fail the handshake.
@@ -142,10 +142,13 @@ class ClientNegotiation {
// Receive a negotiate response message from the server, deserializing it into 'msg'.
// Validates that the response is not an error.
- Status RecvNegotiatePB(NegotiatePB* msg, faststring* buffer) WARN_UNUSED_RESULT;
+ Status RecvNegotiatePB(NegotiatePB* msg,
+ faststring* buffer,
+ std::unique_ptr<ErrorStatusPB>* rpc_error) WARN_UNUSED_RESULT;
// Parse error status message from raw bytes of an ErrorStatusPB.
- Status ParseError(const Slice& err_data) WARN_UNUSED_RESULT;
+ Status ParseError(const Slice& err_data,
+ std::unique_ptr<ErrorStatusPB>* rpc_error) WARN_UNUSED_RESULT;
Status SendConnectionHeader() WARN_UNUSED_RESULT;
@@ -166,11 +169,13 @@ class ClientNegotiation {
// Authenticate to the server using SASL.
// 'recv_buf' allows a receive buffer to be reused.
- Status AuthenticateBySasl(faststring* recv_buf) WARN_UNUSED_RESULT;
+ Status AuthenticateBySasl(faststring* recv_buf,
+ std::unique_ptr<ErrorStatusPB>* rpc_error) WARN_UNUSED_RESULT;
// Authenticate to the server using a token.
// 'recv_buf' allows a receive buffer to be reused.
- Status AuthenticateByToken(faststring* recv_buf) WARN_UNUSED_RESULT;
+ Status AuthenticateByToken(faststring* recv_buf,
+ std::unique_ptr<ErrorStatusPB> *rpc_error) WARN_UNUSED_RESULT;
// Send an SASL_INITIATE message to the server.
// Returns:
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/rpc/connection.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/connection.cc b/src/kudu/rpc/connection.cc
index 912377c..3d3c9f2 100644
--- a/src/kudu/rpc/connection.cc
+++ b/src/kudu/rpc/connection.cc
@@ -140,14 +140,16 @@ bool Connection::Idle() const {
return true;
}
-void Connection::Shutdown(const Status &status) {
+void Connection::Shutdown(const Status &status,
+ unique_ptr<ErrorStatusPB> rpc_error) {
DCHECK(reactor_thread_->IsCurrentThread());
shutdown_status_ = status.CloneAndPrepend("RPC connection failed");
if (inbound_ && inbound_->TransferStarted()) {
double secs_since_active =
(reactor_thread_->cur_time() - last_activity_time_).ToSeconds();
- LOG(WARNING) << "Shutting down " << ToString() << " with pending inbound data ("
+ LOG(WARNING) << "Shutting down " << ToString()
+ << " with pending inbound data ("
<< inbound_->StatusAsString() << ", last active "
<< HumanReadableElapsedTime::ToShortString(secs_since_active)
<< " ago, status=" << status.ToString() << ")";
@@ -157,7 +159,12 @@ void Connection::Shutdown(const Status &status) {
for (const car_map_t::value_type &v : awaiting_response_) {
CallAwaitingResponse *c = v.second;
if (c->call) {
- c->call->SetFailed(status);
+ // Make sure every awaiting call receives the error info, if any.
+ unique_ptr<ErrorStatusPB> error;
+ if (rpc_error) {
+ error.reset(new ErrorStatusPB(*rpc_error));
+ }
+ c->call->SetFailed(status, error.release());
}
// And we must return the CallAwaitingResponse to the pool
car_pool_.Destroy(c);
@@ -638,13 +645,18 @@ std::string Connection::ToString() const {
// regular RPC handling. Destroys Connection on negotiation error.
class NegotiationCompletedTask : public ReactorTask {
public:
- NegotiationCompletedTask(Connection* conn, Status negotiation_status)
+ NegotiationCompletedTask(Connection* conn,
+ Status negotiation_status,
+ std::unique_ptr<ErrorStatusPB> rpc_error)
: conn_(conn),
- negotiation_status_(std::move(negotiation_status)) {
+ negotiation_status_(std::move(negotiation_status)),
+ rpc_error_(std::move(rpc_error)) {
}
virtual void Run(ReactorThread *rthread) OVERRIDE {
- rthread->CompleteConnectionNegotiation(conn_, negotiation_status_);
+ rthread->CompleteConnectionNegotiation(conn_,
+ negotiation_status_,
+ std::move(rpc_error_));
delete this;
}
@@ -658,10 +670,13 @@ class NegotiationCompletedTask : public ReactorTask {
private:
scoped_refptr<Connection> conn_;
const Status negotiation_status_;
+ std::unique_ptr<ErrorStatusPB> rpc_error_;
};
-void Connection::CompleteNegotiation(const Status& negotiation_status) {
- auto task = new NegotiationCompletedTask(this, negotiation_status);
+void Connection::CompleteNegotiation(Status negotiation_status,
+ unique_ptr<ErrorStatusPB> rpc_error) {
+ auto task = new NegotiationCompletedTask(
+ this, std::move(negotiation_status), std::move(rpc_error));
reactor_thread_->reactor()->ScheduleReactorTask(task);
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/rpc/connection.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/connection.h b/src/kudu/rpc/connection.h
index e75b7e6..816c43c 100644
--- a/src/kudu/rpc/connection.h
+++ b/src/kudu/rpc/connection.h
@@ -46,6 +46,7 @@ namespace kudu {
namespace rpc {
class DumpRunningRpcsRequestPB;
+class ErrorStatusPB;
class RpcConnectionPB;
class ReactorThread;
class RpczStore;
@@ -108,7 +109,8 @@ class Connection : public RefCountedThreadSafe<Connection> {
// Fail any calls which are currently queued or awaiting response.
// Prohibits any future calls (they will be failed immediately with this
// same Status).
- void Shutdown(const Status &status);
+ void Shutdown(const Status& status,
+ std::unique_ptr<ErrorStatusPB> rpc_error = {});
// Queue a new call to be made. If the queueing fails, the call will be
// marked failed.
@@ -165,7 +167,8 @@ class Connection : public RefCountedThreadSafe<Connection> {
Socket* socket() { return socket_.get(); }
// Go through the process of transferring control of the underlying socket back to the Reactor.
- void CompleteNegotiation(const Status& negotiation_status);
+ void CompleteNegotiation(Status negotiation_status,
+ std::unique_ptr<ErrorStatusPB> rpc_error);
// Indicate that negotiation is complete and that the Reactor is now in control of the socket.
void MarkNegotiationComplete();
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/rpc/negotiation.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/negotiation.cc b/src/kudu/rpc/negotiation.cc
index 2bc5ea3..66a0112 100644
--- a/src/kudu/rpc/negotiation.cc
+++ b/src/kudu/rpc/negotiation.cc
@@ -17,9 +17,10 @@
#include "kudu/rpc/negotiation.h"
-#include <sys/time.h>
#include <poll.h>
+#include <sys/time.h>
+#include <memory>
#include <ostream>
#include <string>
@@ -65,6 +66,7 @@ DEFINE_bool(rpc_encrypt_loopback_connections, false,
"an attacker.");
TAG_FLAG(rpc_encrypt_loopback_connections, advanced);
+using std::unique_ptr;
using strings::Substitute;
namespace kudu {
@@ -156,7 +158,8 @@ static Status DisableSocketTimeouts(Socket* socket) {
static Status DoClientNegotiation(Connection* conn,
RpcAuthentication authentication,
RpcEncryption encryption,
- MonoTime deadline) {
+ MonoTime deadline,
+ unique_ptr<ErrorStatusPB>* rpc_error) {
const auto* messenger = conn->reactor_thread()->reactor()->messenger();
// Prefer secondary credentials (such as authn token) if permitted by policy.
const auto authn_token = (conn->credentials_policy() == CredentialsPolicy::PRIMARY_CREDENTIALS)
@@ -206,7 +209,7 @@ static Status DoClientNegotiation(Connection* conn,
RETURN_NOT_OK(WaitForClientConnect(client_negotiation.socket(), deadline));
RETURN_NOT_OK(client_negotiation.socket()->SetNonBlocking(false));
- RETURN_NOT_OK(client_negotiation.Negotiate());
+ RETURN_NOT_OK(client_negotiation.Negotiate(rpc_error));
RETURN_NOT_OK(DisableSocketTimeouts(client_negotiation.socket()));
// Transfer the negotiated socket and state back to the connection.
@@ -274,10 +277,12 @@ void Negotiation::RunNegotiation(const scoped_refptr<Connection>& conn,
RpcEncryption encryption,
MonoTime deadline) {
Status s;
+ unique_ptr<ErrorStatusPB> rpc_error;
if (conn->direction() == Connection::SERVER) {
s = DoServerNegotiation(conn.get(), authentication, encryption, deadline);
} else {
- s = DoClientNegotiation(conn.get(), authentication, encryption, deadline);
+ s = DoClientNegotiation(conn.get(), authentication, encryption, deadline,
+ &rpc_error);
}
if (PREDICT_FALSE(!s.ok())) {
@@ -304,7 +309,7 @@ void Negotiation::RunNegotiation(const scoped_refptr<Connection>& conn,
if (conn->direction() == Connection::SERVER && s.IsNotAuthorized()) {
LOG(WARNING) << "Unauthorized connection attempt: " << s.message().ToString();
}
- conn->CompleteNegotiation(s);
+ conn->CompleteNegotiation(std::move(s), std::move(rpc_error));
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/rpc/outbound_call.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/outbound_call.cc b/src/kudu/rpc/outbound_call.cc
index 9b160a1..73025d6 100644
--- a/src/kudu/rpc/outbound_call.cc
+++ b/src/kudu/rpc/outbound_call.cc
@@ -63,7 +63,8 @@ static const double kMicrosPerSecond = 1000000.0;
OutboundCall::OutboundCall(const ConnectionId& conn_id,
const RemoteMethod& remote_method,
google::protobuf::Message* response_storage,
- RpcController* controller, ResponseCallback callback)
+ RpcController* controller,
+ ResponseCallback callback)
: state_(READY),
remote_method_(remote_method),
conn_id_(conn_id),
@@ -282,11 +283,8 @@ void OutboundCall::SetFailed(const Status &status,
{
std::lock_guard<simple_spinlock> l(lock_);
status_ = status;
- if (status_.IsRemoteError()) {
- CHECK(err_pb);
+ if (err_pb) {
error_pb_.reset(err_pb);
- } else {
- CHECK(!err_pb);
}
set_state_unlocked(FINISHED_ERROR);
}
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/rpc/reactor.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/reactor.cc b/src/kudu/rpc/reactor.cc
index 525980b..4141ac3 100644
--- a/src/kudu/rpc/reactor.cc
+++ b/src/kudu/rpc/reactor.cc
@@ -454,11 +454,13 @@ Status ReactorThread::StartConnectionNegotiation(const scoped_refptr<Connection>
return Status::OK();
}
-void ReactorThread::CompleteConnectionNegotiation(const scoped_refptr<Connection>& conn,
- const Status& status) {
+void ReactorThread::CompleteConnectionNegotiation(
+ const scoped_refptr<Connection>& conn,
+ const Status& status,
+ unique_ptr<ErrorStatusPB> rpc_error) {
DCHECK(IsCurrentThread());
if (PREDICT_FALSE(!status.ok())) {
- DestroyConnection(conn.get(), status);
+ DestroyConnection(conn.get(), status, std::move(rpc_error));
return;
}
@@ -466,7 +468,7 @@ void ReactorThread::CompleteConnectionNegotiation(const scoped_refptr<Connection
Status s = conn->SetNonBlocking(true);
if (PREDICT_FALSE(!s.ok())) {
LOG(DFATAL) << "Unable to set connection to non-blocking mode: " << s.ToString();
- DestroyConnection(conn.get(), s);
+ DestroyConnection(conn.get(), s, std::move(rpc_error));
return;
}
@@ -504,10 +506,11 @@ Status ReactorThread::StartConnect(Socket *sock, const Sockaddr& remote) {
}
void ReactorThread::DestroyConnection(Connection *conn,
- const Status& conn_status) {
+ const Status& conn_status,
+ unique_ptr<ErrorStatusPB> rpc_error) {
DCHECK(IsCurrentThread());
- conn->Shutdown(conn_status);
+ conn->Shutdown(conn_status, std::move(rpc_error));
// Unlink connection from lists.
if (conn->direction() == Connection::CLIENT) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/rpc/reactor.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/reactor.h b/src/kudu/rpc/reactor.h
index 615cfc8..37eedd4 100644
--- a/src/kudu/rpc/reactor.h
+++ b/src/kudu/rpc/reactor.h
@@ -185,7 +185,8 @@ class ReactorThread {
// Transition back from negotiating to processing requests.
// Must be called from the reactor thread.
void CompleteConnectionNegotiation(const scoped_refptr<Connection>& conn,
- const Status& status);
+ const Status& status,
+ std::unique_ptr<ErrorStatusPB> rpc_error);
// Collect metrics.
// Must be called from the reactor thread.
@@ -213,7 +214,8 @@ class ReactorThread {
// The connection is not explicitly deleted -- shared_ptr reference counting
// may hold on to the object after this, but callers should assume that it
// _may_ be deleted by this call.
- void DestroyConnection(Connection *conn, const Status &conn_status);
+ void DestroyConnection(Connection *conn, const Status &conn_status,
+ std::unique_ptr<ErrorStatusPB> rpc_error = {});
// Scan any open connections for idle ones that have been idle longer than
// connection_keepalive_time_
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/rpc/retriable_rpc.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/retriable_rpc.h b/src/kudu/rpc/retriable_rpc.h
index ca39abe..a5ff7ad 100644
--- a/src/kudu/rpc/retriable_rpc.h
+++ b/src/kudu/rpc/retriable_rpc.h
@@ -36,16 +36,16 @@ typedef rpc::RequestTracker::SequenceNumber SequenceNumber;
// A base class for retriable RPCs that handles replica picking and retry logic.
//
-// The 'Server' template parameter refers to the the type of the server that will be looked up
+// The 'Server' template parameter refers to the type of the server that will be looked up
// and passed to the derived classes on Try(). For instance in the case of WriteRpc it's
// RemoteTabletServer.
//
-// TODO merge RpcRetrier into this class? Can't be done right now as the retrier is used
+// TODO(unknown): merge RpcRetrier into this class? Can't be done right now as the retrier is used
// independently elsewhere, but likely possible when all replicated RPCs have a ReplicaPicker.
//
-// TODO allow to target replicas other than the leader, if needed.
+// TODO(unknown): allow to target replicas other than the leader, if needed.
//
-// TOOD once we have retry handling on all the RPCs merge this with rpc::Rpc.
+// TODO(unknown): once we have retry handling on all the RPCs merge this with rpc::Rpc.
template <class Server, class RequestPB, class ResponsePB>
class RetriableRpc : public Rpc {
public:
@@ -68,6 +68,11 @@ class RetriableRpc : public Rpc {
// Try() to actually send the request.
void SendRpc() override;
+ // The callback to call upon retrieving (of failing to retrieve) a new authn
+ // token. This is the callback that subclasses should call in their custom
+ // implementation of the GetNewAuthnTokenAndRetry() method.
+ void GetNewAuthnTokenAndRetryCb(const Status& status);
+
protected:
// Subclasses implement this method to actually try the RPC.
// The server been looked up and is ready to be used.
@@ -82,6 +87,15 @@ class RetriableRpc : public Rpc {
// After this is called the RPC will be no longer retried.
virtual void Finish(const Status& status) = 0;
+ // Returns 'true' if the RPC is to scheduled for retry with a new authn token,
+ // 'false' otherwise. For RPCs performed in the context of providing token
+ // for authentication it's necessary to implement this method. The default
+ // implementation returns 'false' meaning the calls returning
+ // INVALID_AUTHENTICATION_TOKEN RPC status are not retried.
+ virtual bool GetNewAuthnTokenAndRetry() {
+ return false;
+ }
+
// Request body.
RequestPB req_;
@@ -90,15 +104,16 @@ class RetriableRpc : public Rpc {
private:
friend class CalculatorServiceRpc;
- // Decides whether to retry the RPC, based on the result of AnalyzeResponse() and retries
- // if that is the case.
+
+ // Decides whether to retry the RPC, based on the result of AnalyzeResponse()
+ // and retries if that is the case.
// Returns true if the RPC was retried or false otherwise.
bool RetryIfNeeded(const RetriableRpcStatus& result, Server* server);
// Called when the replica has been looked up.
void ReplicaFoundCb(const Status& status, Server* server);
- // Called when after the RPC was performed.
+ // Called after the RPC was performed.
void SendRpcCb(const Status& status) override;
// Performs final cleanup, after the RPC is done (independently of success).
@@ -132,8 +147,22 @@ void RetriableRpc<Server, RequestPB, ResponsePB>::SendRpc() {
}
template <class Server, class RequestPB, class ResponsePB>
-bool RetriableRpc<Server, RequestPB, ResponsePB>::RetryIfNeeded(const RetriableRpcStatus& result,
- Server* server) {
+void RetriableRpc<Server, RequestPB, ResponsePB>::GetNewAuthnTokenAndRetryCb(
+ const Status& status) {
+ if (status.ok()) {
+ // Perform the RPC call with the newly fetched authn token.
+ mutable_retrier()->mutable_controller()->Reset();
+ SendRpc();
+ } else {
+ // Back to the retry sequence, hoping for better conditions after some time.
+ VLOG(1) << "Failed to get new authn token: " << status.ToString();
+ mutable_retrier()->DelayedRetry(this, status);
+ }
+}
+
+template <class Server, class RequestPB, class ResponsePB>
+bool RetriableRpc<Server, RequestPB, ResponsePB>::RetryIfNeeded(
+ const RetriableRpcStatus& result, Server* server) {
// Handle the cases where we retry.
switch (result.result) {
case RetriableRpcStatus::SERVICE_UNAVAILABLE:
@@ -141,7 +170,7 @@ bool RetriableRpc<Server, RequestPB, ResponsePB>::RetryIfNeeded(const RetriableR
// SERVICE_UNAVAILABLE error.
break;
- case RetriableRpcStatus::SERVER_NOT_ACCESSIBLE: {
+ case RetriableRpcStatus::SERVER_NOT_ACCESSIBLE:
// TODO(KUDU-1745): not checking for null here results in a crash, since in the case
// of a failed master lookup we have no tablet server corresponding to the error.
//
@@ -153,22 +182,35 @@ bool RetriableRpc<Server, RequestPB, ResponsePB>::RetryIfNeeded(const RetriableR
server_picker_->MarkServerFailed(server, result.status);
}
break;
- }
+
+ case RetriableRpcStatus::RESOURCE_NOT_FOUND:
// The TabletServer was not part of the config serving the tablet.
- // We mark our tablet cache as stale, forcing a master lookup on the next attempt.
- // TODO: Don't backoff the first time we hit this error (see KUDU-1314).
- case RetriableRpcStatus::RESOURCE_NOT_FOUND: {
+ // We mark our tablet cache as stale, forcing a master lookup on the
+ // next attempt.
+ //
+ // TODO(KUDU-1314): Don't backoff the first time we hit this error.
server_picker_->MarkResourceNotFound(server);
-
break;
- }
+
+ case RetriableRpcStatus::REPLICA_NOT_LEADER:
// The TabletServer was not the leader of the quorum.
- case RetriableRpcStatus::REPLICA_NOT_LEADER: {
server_picker_->MarkReplicaNotLeader(server);
break;
+
+ case RetriableRpcStatus::INVALID_AUTHENTICATION_TOKEN: {
+ // This is a special case for retry: first it's necessary to get a new
+ // authn token and then retry the operation with the new token.
+ if (GetNewAuthnTokenAndRetry()) {
+ // The RPC will be retried.
+ resp_.Clear();
+ return true;
+ }
+ // Do not retry.
+ return false;
}
- // For the OK and NON_RETRIABLE_ERROR cases we can't/won't retry.
+
default:
+ // For the OK and NON_RETRIABLE_ERROR cases we can't/won't retry.
return false;
}
resp_.Clear();
@@ -219,7 +261,7 @@ void RetriableRpc<Server, RequestPB, ResponsePB>::SendRpcCb(const Status& status
FinishInternal();
- // From here on out the rpc has either succeeded of suffered a non-retriable
+ // From here on out the RPC has either succeeded of suffered a non-retriable
// failure.
Status final_status = result.status;
if (!final_status.ok()) {
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/rpc/rpc.h
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/rpc.h b/src/kudu/rpc/rpc.h
index a2f41a2..7bc484f 100644
--- a/src/kudu/rpc/rpc.h
+++ b/src/kudu/rpc/rpc.h
@@ -61,6 +61,11 @@ struct RetriableRpcStatus {
// The server doesn't know the resource we're interacting with. For instance a TabletServer
// is not part of the config for the tablet we're trying to write to.
RESOURCE_NOT_FOUND,
+
+ // The authentication token supplied with the operation was found invalid
+ // by the server. Most likely, the token has expired. If so, get a new token
+ // using client credentials and retry the operation with it.
+ INVALID_AUTHENTICATION_TOKEN,
};
Result result;
http://git-wip-us.apache.org/repos/asf/kudu/blob/3db551f7/src/kudu/rpc/server_negotiation.cc
----------------------------------------------------------------------
diff --git a/src/kudu/rpc/server_negotiation.cc b/src/kudu/rpc/server_negotiation.cc
index 7f5e8ee..7cebf8a 100644
--- a/src/kudu/rpc/server_negotiation.cc
+++ b/src/kudu/rpc/server_negotiation.cc
@@ -42,6 +42,8 @@
#include "kudu/security/tls_handshake.h"
#include "kudu/security/tls_socket.h"
#include "kudu/security/token_verifier.h"
+#include "kudu/util/fault_injection.h"
+#include "kudu/util/flag_tags.h"
#include "kudu/util/logging.h"
#include "kudu/util/net/sockaddr.h"
#include "kudu/util/net/socket.h"
@@ -52,8 +54,18 @@ using std::set;
using std::string;
using std::unique_ptr;
+// Fault injection flags.
+DEFINE_double(rpc_inject_invalid_authn_token_ratio, 0,
+ "If set higher than 0, AuthenticateByToken() randomly injects "
+ "errors replying with FATAL_INVALID_AUTHENTICATION_TOKEN code. "
+ "The flag's value corresponds to the probability of the fault "
+ "injection event. Used for only for tests.");
+TAG_FLAG(rpc_inject_invalid_authn_token_ratio, runtime);
+TAG_FLAG(rpc_inject_invalid_authn_token_ratio, unsafe);
+
DECLARE_bool(rpc_encrypt_loopback_connections);
+
namespace kudu {
namespace rpc {
@@ -610,6 +622,33 @@ Status ServerNegotiation::AuthenticateByToken(faststring* recv_buf) {
RETURN_NOT_OK(SendError(ErrorStatusPB::FATAL_INVALID_AUTHENTICATION_TOKEN, s));
return s;
}
+
+ if (PREDICT_FALSE(FLAGS_rpc_inject_invalid_authn_token_ratio > 0)) {
+ security::VerificationResult res;
+ int sel = rand() % 4;
+ switch (sel) {
+ case 0:
+ res = security::VerificationResult::INVALID_TOKEN;
+ break;
+ case 1:
+ res = security::VerificationResult::INVALID_SIGNATURE;
+ break;
+ case 2:
+ res = security::VerificationResult::EXPIRED_TOKEN;
+ break;
+ case 3:
+ res = security::VerificationResult::EXPIRED_SIGNING_KEY;
+ break;
+ }
+ const Status s = kudu::fault_injection::MaybeReturnFailure(
+ FLAGS_rpc_inject_invalid_authn_token_ratio,
+ Status::NotAuthorized(VerificationResultToString(res)));
+ if (!s.ok()) {
+ RETURN_NOT_OK(SendError(ErrorStatusPB::FATAL_INVALID_AUTHENTICATION_TOKEN, s));
+ return s;
+ }
+ }
+
authenticated_user_.SetAuthenticatedByToken(token.authn().username());
// Respond with success message.