You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by fr...@apache.org on 2019/07/02 14:02:17 UTC
[impala] branch master updated (b8e2090 -> aee0abd)
This is an automated email from the ASF dual-hosted git repository.
fredyw pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git.
from b8e2090 Update kudu/security from 9ebcb77aa911aae76c48e717af24e643cb81908d
new 7206b52 IMPALA-6159 / KUDU-2192: Enable TCP keepalive for all outbound connections
new e8c60a3 Update squeasel to 7973705170f4744d1806e32695f7ea1e8308ee95
new aee0abd IMPALA-8716: Log a group of privileges into a single audit event.
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/kudu/rpc/connection.cc | 8 ++
be/src/kudu/rpc/connection.h | 7 ++
be/src/kudu/rpc/reactor.cc | 33 +++++++-
be/src/kudu/rpc/reactor.h | 4 +
be/src/kudu/rpc/rpc-test.cc | 37 ++++++++-
be/src/kudu/util/net/socket.cc | 18 ++++
be/src/kudu/util/net/socket.h | 7 ++
be/src/thirdparty/squeasel/squeasel.c | 63 +++++++++++---
be/src/thirdparty/squeasel/squeasel.h | 19 ++++-
be/src/util/webserver.cc | 21 ++---
be/src/util/webserver.h | 5 +-
.../ranger/RangerAuthorizationChecker.java | 96 ++++++++++++++++------
.../ranger/RangerBufferAuditHandler.java | 10 +++
.../authorization/ranger/RangerAuditLogTest.java | 19 ++++-
14 files changed, 290 insertions(+), 57 deletions(-)
[impala] 01/03: IMPALA-6159 / KUDU-2192: Enable TCP keepalive for
all outbound connections
Posted by fr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fredyw pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 7206b52e5b6ab1df045e4249d859129d32aacf6b
Author: Michael Ho <kw...@cloudera.com>
AuthorDate: Thu Jun 20 18:34:02 2019 -0700
IMPALA-6159 / KUDU-2192: Enable TCP keepalive for all outbound connections
This change enables TCP keepalive for all outbound connections.
This aims to handle cases in which the remote peer may have
dropped off the network without sending a TCP RST. For instance,
a remote host could have hit a kernel panic and got power cycled.
In which case, the existing TCP connection to that host may be
stale. In an idle cluster, this stale connection may not be detected
until the next use of it, in which case it will result in a RPC
failure due to TCP RST sent from the restarted peer.
By enabling TCP keepalive, we ensure that stale TCP connections
in an idle cluster will be detected and closed within a time bound
so a new connection will be created on the next use. This change
introduces 3 different flags:
--tcp_keepalive_probe_period_s: the duration in seconds a TCP connection
has to be idle before keepalive probes started to be sent.
--tcp_keepalive_retry_period_s: the duration in seconds between successive
keepalive probes if previous probes didn't get an ACK from remote peer.
--tcp_keepalive_retry_count: the maximum number of TCP keepalive probes
sent without an ACK before declaring the remote peer as dead.
Testing:
- Used TCP dump to verify that keepalive probes are being sent periodically.
- Verified that blocking all incoming traffic to a server's port via an iptable
rule caused the TCP connection to be closed and the keepalive probes to stop
eventually.
Change-Id: Iaa1d66d83aea1cc82d07fc6217be5fc1306695bc
Reviewed-on: http://gerrit.cloudera.org:8080/13702
Reviewed-by: Alexey Serbin <as...@cloudera.com>
Reviewed-by: Todd Lipcon <to...@apache.org>
Tested-by: Kudu Jenkins
Reviewed-on: http://gerrit.cloudera.org:8080/13764
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/kudu/rpc/connection.cc | 8 ++++++++
be/src/kudu/rpc/connection.h | 7 +++++++
be/src/kudu/rpc/reactor.cc | 33 ++++++++++++++++++++++++++++++++-
be/src/kudu/rpc/reactor.h | 4 ++++
be/src/kudu/rpc/rpc-test.cc | 37 ++++++++++++++++++++++++++++++++++++-
be/src/kudu/util/net/socket.cc | 18 ++++++++++++++++++
be/src/kudu/util/net/socket.h | 7 +++++++
7 files changed, 112 insertions(+), 2 deletions(-)
diff --git a/be/src/kudu/rpc/connection.cc b/be/src/kudu/rpc/connection.cc
index 1632dd3..c9a7576 100644
--- a/be/src/kudu/rpc/connection.cc
+++ b/be/src/kudu/rpc/connection.cc
@@ -82,6 +82,14 @@ Status Connection::SetNonBlocking(bool enabled) {
return socket_->SetNonBlocking(enabled);
}
+Status Connection::SetTcpKeepAlive(int idle_time_s, int retry_time_s, int num_retries) {
+ DCHECK_GT(idle_time_s, 0);
+ DCHECK_GE(retry_time_s, 0);
+ DCHECK_GE(num_retries, 0);
+ return socket_->SetTcpKeepAlive(std::max(1, idle_time_s), std::max(0, retry_time_s),
+ std::max(0, num_retries));
+}
+
void Connection::EpollRegister(ev::loop_ref& loop) {
DCHECK(reactor_thread_->IsCurrentThread());
DVLOG(4) << "Registering connection for epoll: " << ToString();
diff --git a/be/src/kudu/rpc/connection.h b/be/src/kudu/rpc/connection.h
index 362a35b..84b325d 100644
--- a/be/src/kudu/rpc/connection.h
+++ b/be/src/kudu/rpc/connection.h
@@ -98,6 +98,13 @@ class Connection : public RefCountedThreadSafe<Connection> {
// Set underlying socket to non-blocking (or blocking) mode.
Status SetNonBlocking(bool enabled);
+ // Enable TCP keepalive for the underlying socket. A TCP keepalive probe will be sent
+ // to the remote end after the connection has been idle for 'idle_time_s' seconds.
+ // It will retry sending probes up to 'num_retries' number of times until an ACK is
+ // heard from peer. 'retry_time_s' is the sleep time in seconds between successive
+ // keepalive probes.
+ Status SetTcpKeepAlive(int idle_time_s, int retry_time_s, int num_retries);
+
// Register our socket with an epoll loop. We will only ever be registered in
// one epoll loop at a time.
void EpollRegister(ev::loop_ref& loop);
diff --git a/be/src/kudu/rpc/reactor.cc b/be/src/kudu/rpc/reactor.cc
index c1832ef..fd88afb 100644
--- a/be/src/kudu/rpc/reactor.cc
+++ b/be/src/kudu/rpc/reactor.cc
@@ -49,6 +49,7 @@
#include "kudu/util/monotime.h"
#include "kudu/util/net/sockaddr.h"
#include "kudu/util/net/socket.h"
+#include "kudu/util/random_util.h"
#include "kudu/util/status.h"
#include "kudu/util/thread.h"
#include "kudu/util/thread_restrictions.h"
@@ -80,6 +81,21 @@ DEFINE_bool(rpc_reopen_outbound_connections, false,
TAG_FLAG(rpc_reopen_outbound_connections, unsafe);
TAG_FLAG(rpc_reopen_outbound_connections, runtime);
+DEFINE_int32(tcp_keepalive_probe_period_s, 60,
+ "The duration in seconds after an outbound connection has gone idle "
+ "before a TCP keepalive probe is sent to the peer. Set to 0 to disable "
+ "TCP keepalive probes from being sent.");
+DEFINE_int32(tcp_keepalive_retry_period_s, 3,
+ "The duration in seconds between successive keepalive probes from an "
+ "outbound connection if the previous probes are not acknowledged. "
+ "Effective only if --tcp_keepalive_probe_period_s is not 0.");
+DEFINE_int32(tcp_keepalive_retry_count, 10,
+ "The maximum number of keepalive probes sent before declaring the remote "
+ "end as dead. Effective only if --tcp_keepalive_probe_period_s is not 0.");
+TAG_FLAG(tcp_keepalive_probe_period_s, advanced);
+TAG_FLAG(tcp_keepalive_retry_period_s, advanced);
+TAG_FLAG(tcp_keepalive_retry_count, advanced);
+
METRIC_DEFINE_histogram(server, reactor_load_percent,
"Reactor Thread Load Percentage",
kudu::MetricUnit::kUnits,
@@ -133,7 +149,8 @@ ReactorThread::ReactorThread(Reactor *reactor, const MessengerBuilder& bld)
connection_keepalive_time_(bld.connection_keepalive_time_),
coarse_timer_granularity_(bld.coarse_timer_granularity_),
total_client_conns_cnt_(0),
- total_server_conns_cnt_(0) {
+ total_server_conns_cnt_(0),
+ rng_(GetRandomSeed32()) {
if (bld.metric_entity_) {
invoke_us_histogram_ =
@@ -608,6 +625,20 @@ void ReactorThread::CompleteConnectionNegotiation(
return;
}
+ if (FLAGS_tcp_keepalive_probe_period_s > 0) {
+ // Try spreading out the idle poll period to avoid thundering herd in case connections
+ // are all created at the same time (e.g. after a cluster is restarted).
+ Status keepalive_status = conn->SetTcpKeepAlive(
+ FLAGS_tcp_keepalive_probe_period_s + rng_.Uniform32(4),
+ FLAGS_tcp_keepalive_retry_period_s, FLAGS_tcp_keepalive_retry_count);
+ if (PREDICT_FALSE(!keepalive_status.ok())) {
+ LOG(DFATAL) << "Unable to set TCP keepalive for connection: "
+ << keepalive_status.ToString();
+ DestroyConnection(conn.get(), keepalive_status, std::move(rpc_error));
+ return;
+ }
+ }
+
conn->MarkNegotiationComplete();
conn->EpollRegister(loop_);
}
diff --git a/be/src/kudu/rpc/reactor.h b/be/src/kudu/rpc/reactor.h
index ce251c1..e01f71e 100644
--- a/be/src/kudu/rpc/reactor.h
+++ b/be/src/kudu/rpc/reactor.h
@@ -37,6 +37,7 @@
#include "kudu/util/locks.h"
#include "kudu/util/metrics.h"
#include "kudu/util/monotime.h"
+#include "kudu/util/random.h"
#include "kudu/util/status.h"
#include "kudu/util/thread.h"
@@ -327,6 +328,9 @@ class ReactorThread {
// started.
int64_t total_poll_cycles_ = 0;
+ // Random number generator for randomizing the TCP keepalive interval.
+ Random rng_;
+
// Accounting for determining load average in each cycle of TimerHandler.
struct {
// The cycle-time at which the load average was last calculated.
diff --git a/be/src/kudu/rpc/rpc-test.cc b/be/src/kudu/rpc/rpc-test.cc
index 88b12da..5f2be73 100644
--- a/be/src/kudu/rpc/rpc-test.cc
+++ b/be/src/kudu/rpc/rpc-test.cc
@@ -75,6 +75,9 @@ METRIC_DECLARE_histogram(rpc_incoming_queue_time);
DECLARE_bool(rpc_reopen_outbound_connections);
DECLARE_int32(rpc_negotiation_inject_delay_ms);
+DECLARE_int32(tcp_keepalive_probe_period_s);
+DECLARE_int32(tcp_keepalive_retry_period_s);
+DECLARE_int32(tcp_keepalive_retry_count);
using std::shared_ptr;
using std::string;
@@ -766,6 +769,34 @@ TEST_P(TestRpc, TestCallLongerThanKeepalive) {
req, &resp, &controller));
}
+// Test a call which leaves the TCP connection idle for extended period of time
+// and verifies that the call succeeds (i.e. the connection is not closed).
+TEST_P(TestRpc, TestTCPKeepalive) {
+ // Set up server.
+ Sockaddr server_addr;
+ bool enable_ssl = GetParam();
+ ASSERT_OK(StartTestServer(&server_addr, enable_ssl));
+
+ // Set up client.
+ FLAGS_tcp_keepalive_probe_period_s = 1;
+ FLAGS_tcp_keepalive_retry_period_s = 1;
+ FLAGS_tcp_keepalive_retry_count = 1;
+ shared_ptr<Messenger> client_messenger;
+ ASSERT_OK(CreateMessenger("Client", &client_messenger, 1, enable_ssl));
+ Proxy p(client_messenger, server_addr, server_addr.host(),
+ GenericCalculatorService::static_service_name());
+
+ // Make a call which sleeps for longer than TCP keepalive probe period,
+ // triggering TCP keepalive probes.
+ RpcController controller;
+ SleepRequestPB req;
+ req.set_sleep_micros(8 * 1000 * 1000); // 8 seconds.
+ req.set_deferred(true);
+ SleepResponsePB resp;
+ ASSERT_OK(p.SyncRequest(GenericCalculatorService::kSleepMethodName,
+ req, &resp, &controller));
+}
+
// Test that the RpcSidecar transfers the expected messages.
TEST_P(TestRpc, TestRpcSidecar) {
// Set up server.
@@ -1052,11 +1083,15 @@ TEST_F(TestRpc, TestServerShutsDown) {
// EINVAL is possible if the controller socket had already disconnected by
// the time it trys to set the SO_SNDTIMEO socket option as part of the
// normal blocking SASL handshake.
+ //
+ // ENOTCONN is possible simply because the server closes the connection
+ // after the connection is established.
ASSERT_TRUE(s.posix_code() == EPIPE ||
s.posix_code() == ECONNRESET ||
s.posix_code() == ESHUTDOWN ||
s.posix_code() == ECONNREFUSED ||
- s.posix_code() == EINVAL)
+ s.posix_code() == EINVAL ||
+ s.posix_code() == ENOTCONN)
<< "Unexpected status: " << s.ToString();
}
}
diff --git a/be/src/kudu/util/net/socket.cc b/be/src/kudu/util/net/socket.cc
index bc3b525..e58bb4e 100644
--- a/be/src/kudu/util/net/socket.cc
+++ b/be/src/kudu/util/net/socket.cc
@@ -582,6 +582,24 @@ Status Socket::SetTimeout(int opt, const char* optname, const MonoDelta& timeout
return Status::OK();
}
+Status Socket::SetTcpKeepAlive(int idle_time_s, int retry_time_s, int num_retries) {
+#if defined(__linux__)
+ static const char* const err_string = "failed to set socket option $0 to $1";
+ DCHECK_GT(idle_time_s, 0);
+ RETURN_NOT_OK_PREPEND(SetSockOpt(IPPROTO_TCP, TCP_KEEPIDLE, idle_time_s),
+ Substitute(err_string, "TCP_KEEPIDLE", idle_time_s));
+ DCHECK_GT(retry_time_s, 0);
+ RETURN_NOT_OK_PREPEND(SetSockOpt(IPPROTO_TCP, TCP_KEEPINTVL, retry_time_s),
+ Substitute(err_string, "TCP_KEEPINTVL", retry_time_s));
+ DCHECK_GT(num_retries, 0);
+ RETURN_NOT_OK_PREPEND(SetSockOpt(IPPROTO_TCP, TCP_KEEPCNT, num_retries),
+ Substitute(err_string, "TCP_KEEPCNT", num_retries));
+ RETURN_NOT_OK_PREPEND(SetSockOpt(SOL_SOCKET, SO_KEEPALIVE, 1),
+ "failed to enable TCP KeepAlive socket option");
+#endif
+ return Status::OK();
+}
+
template<typename T>
Status Socket::SetSockOpt(int level, int option, const T& value) {
if (::setsockopt(fd_, level, option, &value, sizeof(T)) == -1) {
diff --git a/be/src/kudu/util/net/socket.h b/be/src/kudu/util/net/socket.h
index 992c44a..0f561a7 100644
--- a/be/src/kudu/util/net/socket.h
+++ b/be/src/kudu/util/net/socket.h
@@ -153,6 +153,13 @@ class Socket {
// See also readn() from Stevens (2004) or Kerrisk (2010)
Status BlockingRecv(uint8_t *buf, size_t amt, size_t *nread, const MonoTime& deadline);
+ // Enable TCP keepalive for the underlying socket. A TCP keepalive probe will be sent
+ // to the remote end after the connection has been idle for 'idle_time_s' seconds.
+ // It will retry sending probes up to 'num_retries' number of times until an ACK is
+ // heard from peer. 'retry_time_s' is the sleep time in seconds between successive
+ // keepalive probes.
+ Status SetTcpKeepAlive(int idle_time_s, int retry_time_s, int num_retries);
+
private:
// Called internally from SetSend/RecvTimeout().
Status SetTimeout(int opt, const char* optname, const MonoDelta& timeout);
[impala] 02/03: Update squeasel to
7973705170f4744d1806e32695f7ea1e8308ee95
Posted by fr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fredyw pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit e8c60a30839463f162e31fc76a87de9753531b72
Author: Todd Lipcon <to...@apache.org>
AuthorDate: Fri Jun 28 16:40:07 2019 -0700
Update squeasel to 7973705170f4744d1806e32695f7ea1e8308ee95
This updates to the latest build of squeasel, which fixes a few small
issues and also improves support for keepalive. This patch doesn't
itself enable keepalive, but switches to the new APIs.
Change-Id: I17f90561e0ea6b0917fff51b055225060a4fa549
Reviewed-on: http://gerrit.cloudera.org:8080/13768
Reviewed-by: Todd Lipcon <to...@apache.org>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/thirdparty/squeasel/squeasel.c | 63 +++++++++++++++++++++++++++--------
be/src/thirdparty/squeasel/squeasel.h | 19 ++++++++---
be/src/util/webserver.cc | 21 +++++-------
be/src/util/webserver.h | 5 +--
4 files changed, 77 insertions(+), 31 deletions(-)
diff --git a/be/src/thirdparty/squeasel/squeasel.c b/be/src/thirdparty/squeasel/squeasel.c
index 045740d..b301f5c 100644
--- a/be/src/thirdparty/squeasel/squeasel.c
+++ b/be/src/thirdparty/squeasel/squeasel.c
@@ -923,6 +923,18 @@ static int64_t push(FILE *fp, SOCKET sock, SSL *ssl, const char *buf,
return sent;
}
+// Wait for either 'fd' or 'wakeup_fds' to have readable data.
+static void wait_for_readable_or_wakeup(struct sq_context *ctx,
+ int fd, int timeout_ms) {
+ struct pollfd pfd[2];
+ pfd[0].fd = fd;
+ pfd[0].events = POLLIN;
+ pfd[1].fd = ctx->wakeup_fds[0];
+ pfd[1].events = POLLIN;
+ int poll_rc;
+ RETRY_ON_EINTR(poll_rc, poll(pfd, 2, timeout_ms));
+}
+
// Read from IO channel - opened file descriptor, socket, or SSL descriptor.
// Return negative value on error, or number of bytes read on success.
static int pull(FILE *fp, struct sq_connection *conn, char *buf, int len) {
@@ -939,7 +951,7 @@ static int pull(FILE *fp, struct sq_connection *conn, char *buf, int len) {
#endif
} else {
RETRY_ON_EINTR(nread, recv(conn->client.sock, buf, (size_t) len, 0));
- if (nread == -1) {
+ if (nread == -1 && errno != EAGAIN) {
cry(conn, "error reading: %s", strerror(errno));
}
}
@@ -2535,6 +2547,14 @@ static int read_request(FILE *fp, struct sq_connection *conn,
int request_len, n = 0;
request_len = get_request_len(buf, *nread);
+ if (request_len == 0) {
+ // If we are starting to read a new request, with nothing buffered,
+ // wait for either the beginning of the request, or for the shutdown
+ // signal.
+ wait_for_readable_or_wakeup(conn->ctx, fp ? fileno(fp) : conn->client.sock,
+ atoi(conn->ctx->config[REQUEST_TIMEOUT]));
+ }
+
while (conn->ctx->stop_flag == 0 &&
*nread < bufsiz && request_len == 0 &&
(n = pull(fp, conn, buf + *nread, bufsiz - *nread)) > 0) {
@@ -3850,6 +3870,7 @@ static void handle_request(struct sq_connection *conn) {
char path[PATH_MAX];
int uri_len, ssl_index;
struct file file = STRUCT_FILE_INITIALIZER;
+ sq_callback_result_t callback_result = SQ_HANDLED_OK;
if ((conn->request_info.query_string = strchr(ri->uri, '?')) != NULL) {
* ((char *) conn->request_info.query_string++) = '\0';
@@ -3871,8 +3892,10 @@ static void handle_request(struct sq_connection *conn) {
!check_authorization(conn, path)) {
send_authorization_request(conn);
} else if (conn->ctx->callbacks.begin_request != NULL &&
- conn->ctx->callbacks.begin_request(conn)) {
+ ((callback_result = conn->ctx->callbacks.begin_request(conn))
+ != SQ_CONTINUE_HANDLING)) {
// Do nothing, callback has served the request
+ conn->must_close = (callback_result == SQ_HANDLED_CLOSE_CONNECTION);
#if defined(USE_WEBSOCKET)
} else if (is_websocket_request(conn)) {
handle_websocket_request(conn);
@@ -4313,11 +4336,12 @@ static int set_ssl_option(struct sq_context *ctx) {
EC_KEY* ecdh = EC_KEY_new_by_curve_name(NID_X9_62_prime256v1);
if (ecdh == NULL) {
cry(fc(ctx), "EC_KEY_new_by_curve_name: %s", ssl_error());
- }
-
- int rc = SSL_CTX_set_tmp_ecdh(ctx->ssl_ctx, ecdh);
- if (rc <= 0) {
- cry(fc(ctx), "SSL_CTX_set_tmp_ecdh: %s", ssl_error());
+ } else {
+ int rc = SSL_CTX_set_tmp_ecdh(ctx->ssl_ctx, ecdh);
+ if (rc <= 0) {
+ cry(fc(ctx), "SSL_CTX_set_tmp_ecdh: %s", ssl_error());
+ }
+ EC_KEY_free(ecdh);
}
#elif OPENSSL_VERSION_NUMBER < 0x10100000L
// OpenSSL 1.0.2 provides the set_ecdh_auto API which internally figures out
@@ -4459,7 +4483,14 @@ static int is_valid_uri(const char *uri) {
return uri[0] == '/' || (uri[0] == '*' && uri[1] == '\0');
}
-static int getreq(struct sq_connection *conn, char *ebuf, size_t ebuf_len) {
+
+typedef enum {
+ GETREQ_OK,
+ GETREQ_KEEPALIVE_TIMEOUT,
+ GETREQ_ERROR
+} GetReqResult;
+
+static GetReqResult getreq(struct sq_connection *conn, char *ebuf, size_t ebuf_len) {
const char *cl;
ebuf[0] = '\0';
@@ -4470,11 +4501,13 @@ static int getreq(struct sq_connection *conn, char *ebuf, size_t ebuf_len) {
if (conn->request_len == 0 && conn->data_len == conn->buf_size) {
snprintf(ebuf, ebuf_len, "%s", "Request Too Large");
+ return GETREQ_ERROR;
} else if (conn->request_len <= 0) {
- snprintf(ebuf, ebuf_len, "%s", "Client closed connection");
+ return GETREQ_KEEPALIVE_TIMEOUT;
} else if (parse_http_message(conn->buf, conn->buf_size,
&conn->request_info) <= 0) {
snprintf(ebuf, ebuf_len, "Bad request: [%.*s]", conn->data_len, conn->buf);
+ return GETREQ_ERROR;
} else {
// Request is valid
if ((cl = get_header(&conn->request_info, "Content-Length")) != NULL) {
@@ -4487,7 +4520,7 @@ static int getreq(struct sq_connection *conn, char *ebuf, size_t ebuf_len) {
}
conn->birth_time = time(NULL);
}
- return ebuf[0] == '\0';
+ return GETREQ_OK;
}
struct sq_connection *sq_download(const char *host, int port, int use_ssl,
@@ -4517,6 +4550,7 @@ static void process_new_connection(struct sq_connection *conn) {
struct sq_request_info *ri = &conn->request_info;
int keep_alive_enabled, keep_alive, discard_len;
char ebuf[100];
+ GetReqResult getreq_status;
keep_alive_enabled = !strcmp(conn->ctx->config[ENABLE_KEEP_ALIVE], "yes");
keep_alive = 0;
@@ -4525,8 +4559,11 @@ static void process_new_connection(struct sq_connection *conn) {
// to crule42.
conn->data_len = 0;
do {
- if (!getreq(conn, ebuf, sizeof(ebuf))) {
- send_http_error(conn, 500, "Server Error", "%s", ebuf);
+ getreq_status = getreq(conn, ebuf, sizeof(ebuf));
+ if (getreq_status != GETREQ_OK) {
+ if (getreq_status == GETREQ_ERROR) {
+ send_http_error(conn, 500, "Server Error", "%s", ebuf);
+ }
conn->must_close = 1;
} else if (!is_valid_uri(conn->request_info.uri)) {
char* encoded = (char*) malloc(SQ_BUF_LEN);
@@ -4540,7 +4577,7 @@ static void process_new_connection(struct sq_connection *conn) {
send_http_error(conn, 505, "Bad HTTP version", "%s", ebuf);
}
- if (ebuf[0] == '\0') {
+ if (getreq_status == GETREQ_OK) {
handle_request(conn);
if (conn->ctx->callbacks.end_request != NULL) {
conn->ctx->callbacks.end_request(conn, conn->status_code);
diff --git a/be/src/thirdparty/squeasel/squeasel.h b/be/src/thirdparty/squeasel/squeasel.h
index 68a4034..d4ca155 100644
--- a/be/src/thirdparty/squeasel/squeasel.h
+++ b/be/src/thirdparty/squeasel/squeasel.h
@@ -88,18 +88,29 @@ struct sq_request_info {
} http_headers[64]; // Maximum 64 headers
};
+typedef enum sq_callback_result {
+ // The callback didn't handle the request, and squeasel should
+ // continue with request processing.
+ SQ_CONTINUE_HANDLING = 0,
+ // The callback handled the request, and the connection is still
+ // in a valid state.
+ SQ_HANDLED_OK = 1,
+ // The callback handled the request, but no more requests should
+ // be read from this connection (eg the request was invalid).
+ SQ_HANDLED_CLOSE_CONNECTION = 2
+} sq_callback_result_t;
// This structure needs to be passed to sq_start(), to let squeasel know
// which callbacks to invoke. For detailed description, see
// https://github.com/cloudera/squeasel/blob/master/UserManual.md
struct sq_callbacks {
// Called when squeasel has received new HTTP request.
- // If callback returns non-zero,
+ // If callback returns one of the SQ_HANDLED_* results,
// callback must process the request by sending valid HTTP headers and body,
// and squeasel will not do any further processing.
- // If callback returns 0, squeasel processes the request itself. In this case,
- // callback must not send any data to the client.
- int (*begin_request)(struct sq_connection *);
+ // If callback returns SQ_CONTINUE_HANDLING, squeasel processes the request itself.
+ // In this case, callback must not send any data to the client.
+ sq_callback_result_t (*begin_request)(struct sq_connection *);
// Called when squeasel has finished processing request.
void (*end_request)(const struct sq_connection *, int reply_status_code);
diff --git a/be/src/util/webserver.cc b/be/src/util/webserver.cc
index 3ea3e65..caf45d3 100644
--- a/be/src/util/webserver.cc
+++ b/be/src/util/webserver.cc
@@ -112,10 +112,6 @@ DECLARE_string(ssl_cipher_list);
static const char* DOC_FOLDER = "/www/";
static const int DOC_FOLDER_LEN = strlen(DOC_FOLDER);
-// Easy-to-read constants for Squeasel return codes
-static const uint32_t PROCESSING_COMPLETE = 1;
-static const uint32_t NOT_PROCESSED = 0;
-
// Standard key in the json document sent to templates for rendering. Must be kept in
// sync with the templates themselves.
static const char* COMMON_JSON_KEY = "__common__";
@@ -391,23 +387,24 @@ int Webserver::LogMessageCallbackStatic(const struct sq_connection* connection,
if (message != nullptr) {
LOG(INFO) << "Webserver: " << message;
}
- return PROCESSING_COMPLETE;
+ return SQ_HANDLED_OK;
}
-int Webserver::BeginRequestCallbackStatic(struct sq_connection* connection) {
+sq_callback_result_t Webserver::BeginRequestCallbackStatic(
+ struct sq_connection* connection) {
struct sq_request_info* request_info = sq_get_request_info(connection);
Webserver* instance = reinterpret_cast<Webserver*>(request_info->user_data);
return instance->BeginRequestCallback(connection, request_info);
}
-int Webserver::BeginRequestCallback(struct sq_connection* connection,
+sq_callback_result_t Webserver::BeginRequestCallback(struct sq_connection* connection,
struct sq_request_info* request_info) {
if (!FLAGS_webserver_doc_root.empty() && FLAGS_enable_webserver_doc_root) {
if (strncmp(DOC_FOLDER, request_info->uri, DOC_FOLDER_LEN) == 0) {
VLOG(2) << "HTTP File access: " << request_info->uri;
// Let Squeasel deal with this request; returning NULL will fall through
// to the default handler which will serve files.
- return NOT_PROCESSED;
+ return SQ_CONTINUE_HANDLING;
}
}
@@ -445,7 +442,7 @@ int Webserver::BeginRequestCallback(struct sq_connection* connection,
sq_printf(connection,
"HTTP/1.1 %s\r\n",
HttpStatusCodeToString(HttpStatusCode::LengthRequired).c_str());
- return 1;
+ return SQ_HANDLED_OK;
}
if (content_len > FLAGS_webserver_max_post_length_bytes) {
// TODO: for this and other HTTP requests, we should log the
@@ -454,7 +451,7 @@ int Webserver::BeginRequestCallback(struct sq_connection* connection,
sq_printf(connection,
"HTTP/1.1 %s\r\n",
HttpStatusCodeToString(HttpStatusCode::RequestEntityTooLarge).c_str());
- return 1;
+ return SQ_HANDLED_CLOSE_CONNECTION;
}
char buf[8192];
@@ -468,7 +465,7 @@ int Webserver::BeginRequestCallback(struct sq_connection* connection,
sq_printf(connection,
"HTTP/1.1 %s\r\n",
HttpStatusCodeToString(HttpStatusCode::InternalServerError).c_str());
- return 1;
+ return SQ_HANDLED_CLOSE_CONNECTION;
}
req.post_data.append(buf, n);
@@ -501,7 +498,7 @@ int Webserver::BeginRequestCallback(struct sq_connection* connection,
// Make sure to use sq_write for printing the body; sq_printf truncates at 8kb
sq_write(connection, str.c_str(), str.length());
- return PROCESSING_COMPLETE;
+ return SQ_HANDLED_OK;
}
void Webserver::RenderUrlWithTemplate(const WebRequest& req,
diff --git a/be/src/util/webserver.h b/be/src/util/webserver.h
index 2fbc0cd..21b85be 100644
--- a/be/src/util/webserver.h
+++ b/be/src/util/webserver.h
@@ -156,10 +156,11 @@ class Webserver {
/// Squeasel callback for HTTP request events. Static so that it can act as a function
/// pointer, and then call the next method. Returns squeasel success code.
- static int BeginRequestCallbackStatic(struct sq_connection* connection);
+ static sq_callback_result_t BeginRequestCallbackStatic(
+ struct sq_connection* connection);
/// Dispatch point for all incoming requests. Returns squeasel success code.
- int BeginRequestCallback(struct sq_connection* connection,
+ sq_callback_result_t BeginRequestCallback(struct sq_connection* connection,
struct sq_request_info* request_info);
/// Renders URLs through the Mustache templating library.
[impala] 03/03: IMPALA-8716: Log a group of privileges into a
single audit event.
Posted by fr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
fredyw pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit aee0abd76b762e57ce9f0a2e40a9a8b4f97dc986
Author: Fredy Wijaya <fw...@cloudera.com>
AuthorDate: Wed Jun 26 16:30:50 2019 -0700
IMPALA-8716: Log a group of privileges into a single audit event.
This patch updates the audit log handler to group a privilege that
consists of multiple privileges into a single audit event.
For example if we run "show partitions foo.bar" and we have
SELECT privilege on table "foo.bar", before this patch, we would be
creating 2 audit events:
- Attempt to check if there's INSERT privilege on table "foo.bar"
Result: denied, access type: insert, resource: foo.bar
- Attempt to check if there's SELECT privilege on table "foo.bar"
Result: allowed, access type: select, resource: foo.bar
After this patch, we will only create a single audit event, e.g.
Result: allowed, access type: view_metadata, resource: foo.bar
Testing:
- Updated tests in RangerAuditLogTest
- Ran FE tests
Change-Id: Ib53102bc1ceaf9d62544090dc00f3231fae0efca
Reviewed-on: http://gerrit.cloudera.org:8080/13744
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../ranger/RangerAuthorizationChecker.java | 96 ++++++++++++++++------
.../ranger/RangerBufferAuditHandler.java | 10 +++
.../authorization/ranger/RangerAuditLogTest.java | 19 ++++-
3 files changed, 101 insertions(+), 24 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/authorization/ranger/RangerAuthorizationChecker.java b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerAuthorizationChecker.java
index 7f76cad..9a4e67d 100644
--- a/fe/src/main/java/org/apache/impala/authorization/ranger/RangerAuthorizationChecker.java
+++ b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerAuthorizationChecker.java
@@ -151,15 +151,14 @@ public class RangerAuthorizationChecker extends BaseAuthorizationChecker {
for (RangerAccessResourceImpl resource: resources) {
if (request.getPrivilege() == Privilege.ANY) {
- if (!authorizeResource(rangerAuthzCtx, resource, user, request.getPrivilege())) {
+ if (!authorizeResource(rangerAuthzCtx, resource, user, request.getPrivilege(),
+ ((RangerAuthorizationContext) authzCtx).getAuditHandler())) {
return false;
}
} else {
boolean authorized = request.getPrivilege().hasAnyOf() ?
- authorizeAny(rangerAuthzCtx, resource, user,
- request.getPrivilege().getImpliedPrivileges()) :
- authorizeAll(rangerAuthzCtx, resource, user,
- request.getPrivilege().getImpliedPrivileges());
+ authorizeAny(rangerAuthzCtx, resource, user, request.getPrivilege()) :
+ authorizeAll(rangerAuthzCtx, resource, user, request.getPrivilege());
if (!authorized) {
return false;
}
@@ -240,9 +239,7 @@ public class RangerAuthorizationChecker extends BaseAuthorizationChecker {
// event
RangerAuthorizationContext tmpCtx = new RangerAuthorizationContext(
originalCtx.getSessionState(), originalCtx.getTimeline());
- tmpCtx.setAuditHandler(new RangerBufferAuditHandler(
- originalAuditHandler.getSqlStmt(), originalAuditHandler.getClusterName(),
- originalAuditHandler.getClientIp()));
+ tmpCtx.setAuditHandler(new RangerBufferAuditHandler(originalAuditHandler));
try {
super.authorizeTableAccess(tmpCtx, analysisResult, catalog, requests);
} catch (AuthorizationException e) {
@@ -314,30 +311,84 @@ public class RangerAuthorizationChecker extends BaseAuthorizationChecker {
}
private boolean authorizeAny(RangerAuthorizationContext authzCtx,
- RangerAccessResourceImpl resource, User user, EnumSet<Privilege> privileges)
+ RangerAccessResourceImpl resource, User user, Privilege privilege)
throws InternalException {
- for (Privilege privilege: privileges) {
- if (authorizeResource(authzCtx, resource, user, privilege)) {
- return true;
+ boolean authorized = false;
+ RangerBufferAuditHandler originalAuditHandler = authzCtx.getAuditHandler();
+ // Use a temporary audit handler instead of the original audit handler
+ // so that we know all the audit events generated by the temporary audit
+ // handler are contained to the given resource.
+ RangerBufferAuditHandler tmpAuditHandler = originalAuditHandler == null ?
+ null : new RangerBufferAuditHandler(originalAuditHandler);
+ for (Privilege impliedPrivilege: privilege.getImpliedPrivileges()) {
+ if (authorizeResource(authzCtx, resource, user, impliedPrivilege,
+ tmpAuditHandler)) {
+ authorized = true;
+ break;
}
}
- return false;
+ if (originalAuditHandler != null) {
+ updateAuditEvents(tmpAuditHandler, originalAuditHandler, true /*any*/,
+ privilege);
+ }
+ return authorized;
}
private boolean authorizeAll(RangerAuthorizationContext authzCtx,
- RangerAccessResourceImpl resource, User user, EnumSet<Privilege> privileges)
+ RangerAccessResourceImpl resource, User user, Privilege privilege)
throws InternalException {
- for (Privilege privilege: privileges) {
- if (!authorizeResource(authzCtx, resource, user, privilege)) {
- return false;
+ boolean authorized = true;
+ RangerBufferAuditHandler originalAuditHandler = authzCtx.getAuditHandler();
+ // Use a temporary audit handler instead of the original audit handler
+ // so that we know all the audit events generated by the temporary audit
+ // handler are contained to the given resource.
+ RangerBufferAuditHandler tmpAuditHandler = originalAuditHandler == null ?
+ null : new RangerBufferAuditHandler(originalAuditHandler);
+ for (Privilege impliedPrivilege: privilege.getImpliedPrivileges()) {
+ if (!authorizeResource(authzCtx, resource, user, impliedPrivilege,
+ tmpAuditHandler)) {
+ authorized = false;
+ break;
}
}
- return true;
+ if (originalAuditHandler != null) {
+ updateAuditEvents(tmpAuditHandler, originalAuditHandler, false /*not any*/,
+ privilege);
+ }
+ return authorized;
+ }
+
+ /**
+ * Updates the {@link AuthzAuditEvent} from the source to the destination
+ * {@link RangerBufferAuditHandler} with the given privilege.
+ *
+ * For example we want to log:
+ * VIEW_METADATA - Allowed
+ * vs
+ * INSERT - Denied, INSERT - Allowed.
+ */
+ private static void updateAuditEvents(RangerBufferAuditHandler source,
+ RangerBufferAuditHandler dest, boolean isAny, Privilege privilege) {
+ // A custom audit event to log the actual privilege instead of the implied
+ // privileges.
+ AuthzAuditEvent newAuditEvent = null;
+ for (AuthzAuditEvent auditEvent : source.getAuthzEvents()) {
+ if (auditEvent.getAccessResult() == (isAny ? 1 : 0)) {
+ newAuditEvent = auditEvent;
+ break; // break if at least one is a success.
+ } else {
+ newAuditEvent = auditEvent;
+ }
+ }
+ if (newAuditEvent != null) {
+ newAuditEvent.setAccessType(privilege.name().toLowerCase());
+ dest.getAuthzEvents().add(newAuditEvent);
+ }
}
private boolean authorizeResource(RangerAuthorizationContext authzCtx,
- RangerAccessResourceImpl resource, User user, Privilege privilege)
- throws InternalException {
+ RangerAccessResourceImpl resource, User user, Privilege privilege,
+ RangerBufferAuditHandler auditHandler) throws InternalException {
String accessType;
if (privilege == Privilege.ANY) {
accessType = RangerPolicyEngine.ANY_ACCESS;
@@ -354,9 +405,8 @@ public class RangerAuthorizationChecker extends BaseAuthorizationChecker {
request.setClientIPAddress(
authzCtx.getSessionState().getNetwork_address().getHostname());
}
- RangerAccessResult result = plugin_.isAccessAllowed(request,
- authzCtx.getAuditHandler());
- return result != null && result.getIsAllowed();
+ RangerAccessResult authorized = plugin_.isAccessAllowed(request, auditHandler);
+ return authorized != null && authorized.getIsAllowed();
}
@VisibleForTesting
diff --git a/fe/src/main/java/org/apache/impala/authorization/ranger/RangerBufferAuditHandler.java b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerBufferAuditHandler.java
index 4b192c5..b922080 100644
--- a/fe/src/main/java/org/apache/impala/authorization/ranger/RangerBufferAuditHandler.java
+++ b/fe/src/main/java/org/apache/impala/authorization/ranger/RangerBufferAuditHandler.java
@@ -61,6 +61,16 @@ public class RangerBufferAuditHandler implements RangerAccessResultProcessor {
clientIp_ = Preconditions.checkNotNull(clientIp);
}
+ /**
+ * A copy constructor for {@link RangerBufferAuditHandler}.
+ */
+ public RangerBufferAuditHandler(RangerBufferAuditHandler auditHandler) {
+ Preconditions.checkNotNull(auditHandler);
+ sqlStmt_ = Preconditions.checkNotNull(auditHandler.getSqlStmt());
+ clusterName_ = Preconditions.checkNotNull(auditHandler.getClusterName());
+ clientIp_ = Preconditions.checkNotNull(auditHandler.getClientIp());
+ }
+
public static class AutoFlush extends RangerBufferAuditHandler
implements AutoCloseable {
public AutoFlush(String sqlStmt, String clusterName, String clientIp) {
diff --git a/fe/src/test/java/org/apache/impala/authorization/ranger/RangerAuditLogTest.java b/fe/src/test/java/org/apache/impala/authorization/ranger/RangerAuditLogTest.java
index fb85d7b..1f4a6a4 100644
--- a/fe/src/test/java/org/apache/impala/authorization/ranger/RangerAuditLogTest.java
+++ b/fe/src/test/java/org/apache/impala/authorization/ranger/RangerAuditLogTest.java
@@ -134,6 +134,15 @@ public class RangerAuditLogTest extends AuthorizationTestBase {
assertEventEquals("@url", "refresh", "*", 1, events.get(2));
assertEquals("invalidate metadata", events.get(2).getRequestData());
}, "invalidate metadata", onServer(TPrivilegeLevel.REFRESH));
+
+ authzOk(events -> {
+ // Show the actual view_metadata audit log.
+ assertEquals(1, events.size());
+ assertEventEquals("@table", "view_metadata", "functional/alltypes", 1,
+ events.get(0));
+ assertEquals("show partitions functional.alltypes", events.get(0).getRequestData());
+ }, "show partitions functional.alltypes",
+ onTable("functional", "alltypes", TPrivilegeLevel.SELECT));
}
@Test
@@ -180,6 +189,14 @@ public class RangerAuditLogTest extends AuthorizationTestBase {
assertEquals("select id, string_col from functional.alltypes",
events.get(0).getRequestData());
}, "select id, string_col from functional.alltypes");
+
+ authzError(events -> {
+ // Show the actual view_metadata audit log.
+ assertEquals(1, events.size());
+ assertEventEquals("@table", "view_metadata", "functional/alltypes", 0,
+ events.get(0));
+ assertEquals("show partitions functional.alltypes", events.get(0).getRequestData());
+ }, "show partitions functional.alltypes");
}
private void authzOk(Consumer<List<AuthzAuditEvent>> resultChecker, String stmt,
@@ -201,7 +218,7 @@ public class RangerAuditLogTest extends AuthorizationTestBase {
private static void assertEventEquals(String resourceType, String accessType,
String resourcePath, int accessResult, AuthzAuditEvent event) {
assertEquals(resourceType, event.getResourceType());
- assertEquals(accessType.toUpperCase(), event.getAccessType());
+ assertEquals(accessType, event.getAccessType());
assertEquals(resourcePath, event.getResourcePath());
assertEquals(accessResult, event.getAccessResult());
assertEquals("test-cluster", event.getClusterName());