You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by aw...@apache.org on 2020/03/30 18:53:33 UTC
[kudu] branch master updated (4e24702 -> 5cd26cf)
This is an automated email from the ASF dual-hosted git repository.
awong pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git.
from 4e24702 [build-support] introduce PARALLEL variable
new b9faf6c [ranger] add more comments for Ranger privilege model
new 5660d95 [ranger] fix authorizing list tables
new 697301f subprocess: use a fifo instead of stdout for IO
new 5cd26cf env: add a fifo class
The 4 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:
.../kudu/subprocess/SubprocessConfiguration.java | 46 +++++++++++
.../apache/kudu/subprocess/SubprocessExecutor.java | 4 +-
src/kudu/master/catalog_manager.cc | 3 +-
src/kudu/master/ranger_authz_provider.cc | 7 +-
src/kudu/master/ranger_authz_provider.h | 11 +--
src/kudu/ranger/ranger.proto | 37 ++++++---
src/kudu/ranger/ranger_client-test.cc | 11 ++-
src/kudu/ranger/ranger_client.cc | 47 +++++++----
src/kudu/ranger/ranger_client.h | 14 +++-
src/kudu/subprocess/server.cc | 34 +++++++-
src/kudu/subprocess/server.h | 15 +++-
src/kudu/subprocess/subprocess_proxy-test.cc | 12 ++-
src/kudu/subprocess/subprocess_proxy.h | 6 +-
src/kudu/subprocess/subprocess_server-test.cc | 81 ++++++++++++++++---
src/kudu/util/env-test.cc | 32 ++++++++
src/kudu/util/env.h | 27 ++++++-
src/kudu/util/env_posix.cc | 94 +++++++++++++++++-----
17 files changed, 393 insertions(+), 88 deletions(-)
[kudu] 02/04: [ranger] fix authorizing list tables
Posted by aw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
awong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 5660d9583e88c4ece2c73efc2cfcb51bd3d22523
Author: Hao Hao <ha...@cloudera.com>
AuthorDate: Mon Mar 30 00:03:38 2020 -0700
[ranger] fix authorizing list tables
When authorizing list tables and there is no tables in Kudu, we should
return 'OK' instead of 'InvalidArgument' error.
Change-Id: I2a4dfe61e85fc21d67e6422a1cdb36d887d51f4f
Reviewed-on: http://gerrit.cloudera.org:8080/15592
Reviewed-by: Attila Bukor <ab...@apache.org>
Reviewed-by: Andrew Wong <aw...@cloudera.com>
Tested-by: Kudu Jenkins
---
src/kudu/ranger/ranger_client-test.cc | 6 ++++++
src/kudu/ranger/ranger_client.cc | 3 ++-
2 files changed, 8 insertions(+), 1 deletion(-)
diff --git a/src/kudu/ranger/ranger_client-test.cc b/src/kudu/ranger/ranger_client-test.cc
index b2cae2c..9a85aaa 100644
--- a/src/kudu/ranger/ranger_client-test.cc
+++ b/src/kudu/ranger/ranger_client-test.cc
@@ -161,6 +161,12 @@ TEST_F(RangerClientTest, TestAuthorizeCreateTableAuthorized) {
TEST_F(RangerClientTest, TestAuthorizeListNoTables) {
unordered_set<string> tables;
+ ASSERT_OK(client_.AuthorizeActionMultipleTables("jdoe", ActionPB::METADATA, &tables));
+ ASSERT_EQ(0, tables.size());
+}
+
+TEST_F(RangerClientTest, TestAuthorizeListNoTablesAuthorized) {
+ unordered_set<string> tables;
tables.emplace("foo.bar");
tables.emplace("foo.baz");
auto s = client_.AuthorizeActionMultipleTables("jdoe", ActionPB::METADATA, &tables);
diff --git a/src/kudu/ranger/ranger_client.cc b/src/kudu/ranger/ranger_client.cc
index c67e03c..a2abee1 100644
--- a/src/kudu/ranger/ranger_client.cc
+++ b/src/kudu/ranger/ranger_client.cc
@@ -334,8 +334,9 @@ Status RangerClient::AuthorizeActionMultipleTables(const string& user_name,
const ActionPB& action,
unordered_set<string>* table_names) {
DCHECK(subprocess_);
+ // Return immediately if there is no tables to authorize against.
if (table_names->empty()) {
- return Status::InvalidArgument("Empty set of tables");
+ return Status::OK();
}
RangerRequestListPB req_list;
[kudu] 01/04: [ranger] add more comments for Ranger privilege model
Posted by aw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
awong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit b9faf6c23ac1499bddf7b32b6433268d12df642b
Author: Hao Hao <ha...@cloudera.com>
AuthorDate: Thu Mar 26 15:57:52 2020 -0700
[ranger] add more comments for Ranger privilege model
Change-Id: I1b827d06c7657e450903333d820dc9ef9313dcd6
Reviewed-on: http://gerrit.cloudera.org:8080/15566
Tested-by: Kudu Jenkins
Reviewed-by: Andrew Wong <aw...@cloudera.com>
---
src/kudu/master/ranger_authz_provider.h | 8 -------
src/kudu/ranger/ranger.proto | 37 +++++++++++++++++++++++----------
src/kudu/ranger/ranger_client.h | 7 ++++++-
3 files changed, 32 insertions(+), 20 deletions(-)
diff --git a/src/kudu/master/ranger_authz_provider.h b/src/kudu/master/ranger_authz_provider.h
index a97ad12..bdfb11a 100644
--- a/src/kudu/master/ranger_authz_provider.h
+++ b/src/kudu/master/ranger_authz_provider.h
@@ -40,14 +40,6 @@ namespace master {
// An implementation of AuthzProvider that connects to Ranger and translates
// authorization requests to Ranger and allows or denies the actions based on
// the received responses.
-//
-// The privilege model for Kudu operations with Ranger follows the existing
-// one enforced with Sentry (see sentry_authz_provider.cc). However note that
-// in terms of policy evaluation, Ranger is different than Sentry that a policy
-// with a higher scope in the hierarchy cannot imply a lower scope its hierarchy
-// tree. e.g. 'METADATA on db=a' cannot imply 'METADATA on db=a->table=tbl'.
-// Therefore, in Ranger world one can grant 'METADATA on db=a->table=*->column=*'
-// to match with Sentry policy 'METADATA on db=a'.
class RangerAuthzProvider : public AuthzProvider {
public:
diff --git a/src/kudu/ranger/ranger.proto b/src/kudu/ranger/ranger.proto
index 939f3b0..1f9169a 100644
--- a/src/kudu/ranger/ranger.proto
+++ b/src/kudu/ranger/ranger.proto
@@ -18,17 +18,32 @@ syntax = "proto2";
package kudu.ranger;
option java_package = "org.apache.kudu.ranger";
-// Describes the type of action that can be performed in Ranger.
-//
-// SQL-like action types used by Ranger. ALL implies all other privilege types
-// and all privilege types imply METADATA. METADATA is used for discovery
-// (listing tables).
-//
-// The action type mapping is similar to the one in Sentry which was implemented
-// before Ranger and the same privileges have to be enforced with both
-// authorization providers.
-//
-// For more information on fine grained authz check out docs/security.adoc
+// Similar to Sentry, in Ranger, an action is an operation taken on an
+// authorizable, an authorizable is a linear hierarchically structured
+// resource (database -> table -> column), and 'privileges' are composed
+// of an authorizable and an action, e.g. CREATE ON DATABASE a (db=a).
+
+// SQL-like action types are used in Ranger. All actions are independent,
+// except that ALL subsumes every other action, and every action subsumes
+// METADATA.
+
+// In term of privilege evaluation, unlike Sentry, (where authorizables higher
+// up on the hierarchy can imply authorizables lower on the hierarchy, e.g.
+// database implies table), Ranger doesn't have the concept of hierarchical
+// implication. To be more specific, privilege 'METADATA ON DATABASE a (db=a)'
+// does not imply 'METADATA ON TABLE a.tbl (db=a->table=tbl)'. Thus, in Ranger
+// users granted with 'METADATA ON DATABASE a' cannot perform an action that
+// requires 'METADATA ON TABLE a.tbl'. On the other hand, Ranger supports
+// wildcard matching on authorizables, e.g. 'db=a->table=*' matches all the
+// tables that belong to DATABASE a. Therefore, in Ranger users actually need
+// 'METADATA ON db=a->table=*->column=*' privilege to match the semantics of
+// 'METADATA ON db=a' in Sentry.
+
+// Nevertheless, the same set of privileges are enforced/required for Kudu
+// operations with both Sentry and Ranger. For the detailed privilege
+// enforcement information see 'Policy for Kudu Masters/Tablet Servers'
+// section in docs/security.adoc.
+
enum ActionPB {
SELECT = 0;
INSERT = 1;
diff --git a/src/kudu/ranger/ranger_client.h b/src/kudu/ranger/ranger_client.h
index 2b1bbe1..cdfb8ed 100644
--- a/src/kudu/ranger/ranger_client.h
+++ b/src/kudu/ranger/ranger_client.h
@@ -52,7 +52,12 @@ typedef subprocess::SubprocessProxy<RangerRequestListPB, RangerResponseListPB,
class RangerClient {
public:
// Similar to SentryAuthorizableScope scope which indicates the
- // hierarchy of authorizables (database → table).
+ // hierarchy of authorizables (database -> table -> column). For
+ // example, authorizable 'db=a' has database level scope, while
+ // authorizable 'db=a->table=b' has table level scope. Note that
+ // COLUMN level scope is not defined in the enum as it is not
+ // used yet in the code (although the concept still apply when
+ // authorizing column level privileges).
enum Scope {
DATABASE,
TABLE
[kudu] 03/04: subprocess: use a fifo instead of stdout for IO
Posted by aw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
awong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 697301fb076e9649e7f8eb4556c91e59ab9b185f
Author: Andrew Wong <aw...@cloudera.com>
AuthorDate: Fri Mar 27 22:21:21 2020 -0700
subprocess: use a fifo instead of stdout for IO
Today, communication from a Kudu Java subprocess to the C++
SubprocessServer is done via the subprocess's stdout (the C++ message
protocol listens on the stdout FD of the subprocess). This is risky
because any amount of logging by the subprocess to stdout may end up
breaking the message protocol, preventing any calls to the subprocess
from succeeding.
This updates the Kudu subprocess module on both the Java and C++ side to
use a fifo for output instead of the subprocess's stdout. For Ranger,
this fifo will by default be placed in the Ranger config directory,
which, in some distributions (e.g. CDPD) will be unique per Kudu
process. Additionally, the name of this file is chosen to avoid
collisions with existing fifos.
This patch also updates the subprocess to share stdout and stderr with
the parent process by default, adding basic log4j configuration to the
Java subprocess to output to System.out. Here's a snippet of the output
from subprocess_proxy-test (which includes both Kudu's glog and stdout):
0 [pool-1-thread-1] DEBUG org.apache.kudu.subprocess.QueueUtil - Message: org.apache.kudu.subprocess.InboundRequest@48c4ccb9 has been put on the queue
0 [pool-2-thread-1] DEBUG org.apache.kudu.subprocess.QueueUtil - Message: org.apache.kudu.subprocess.InboundRequest@48c4ccb9 has been taken from the queue
1096 [pool-2-thread-1] DEBUG org.apache.kudu.subprocess.QueueUtil - Message: org.apache.kudu.subprocess.OutboundResponse@6c275a0f has been put on the queue
1096 [pool-3-thread-1] DEBUG org.apache.kudu.subprocess.QueueUtil - Message: org.apache.kudu.subprocess.OutboundResponse@6c275a0f has been taken from the queue
WARNING: Logging before InitGoogleLogging() is written to STDERR
I0327 23:35:20.583730 244412416 server.cc:279] Received an EOF from the subprocess
I0327 23:35:20.590333 244948992 server.cc:412] outbound queue shut down: Aborted:
This may not be the final form of our logging story for the subprocess
(e.g. maybe we'll want to log to a file, or maybe we'll want to try
folding it into the server's glog). But at the very least, this is
better than not having logs at all, which is the state of logging today.
I considered using Netty's FileDescriptor[1] to leverage the FDs from
calls to pipe(). I opted not to since that brought in some hairiness
around its usage of JNI (I saw an UnsatisfiedLinkError when I tried
writing to the FD).
[1] https://netty.io/4.0/api/io/netty/channel/unix/FileDescriptor.html
Change-Id: I63e91a090fe196713a013d28301c7980e452456c
Reviewed-on: http://gerrit.cloudera.org:8080/15574
Tested-by: Kudu Jenkins
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Reviewed-by: Attila Bukor <ab...@apache.org>
Reviewed-by: Hao Hao <ha...@cloudera.com>
---
.../kudu/subprocess/SubprocessConfiguration.java | 46 ++++++++++++
.../apache/kudu/subprocess/SubprocessExecutor.java | 4 +-
src/kudu/master/catalog_manager.cc | 3 +-
src/kudu/master/ranger_authz_provider.cc | 7 +-
src/kudu/master/ranger_authz_provider.h | 3 +-
src/kudu/ranger/ranger_client-test.cc | 5 +-
src/kudu/ranger/ranger_client.cc | 44 ++++++++----
src/kudu/ranger/ranger_client.h | 7 +-
src/kudu/subprocess/server.cc | 34 +++++++--
src/kudu/subprocess/server.h | 15 +++-
src/kudu/subprocess/subprocess_proxy-test.cc | 12 +++-
src/kudu/subprocess/subprocess_proxy.h | 6 +-
src/kudu/subprocess/subprocess_server-test.cc | 81 ++++++++++++++++++----
13 files changed, 221 insertions(+), 46 deletions(-)
diff --git a/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/SubprocessConfiguration.java b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/SubprocessConfiguration.java
index 7234695..e80f3ba 100644
--- a/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/SubprocessConfiguration.java
+++ b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/SubprocessConfiguration.java
@@ -17,6 +17,13 @@
package org.apache.kudu.subprocess;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.RandomAccessFile;
+
import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.cli.BasicParser;
import org.apache.commons.cli.CommandLine;
@@ -24,6 +31,7 @@ import org.apache.commons.cli.CommandLineParser;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.apache.commons.cli.ParseException;
+import org.apache.log4j.BasicConfigurator;
import org.apache.yetus.audience.InterfaceAudience;
/**
@@ -40,6 +48,7 @@ public class SubprocessConfiguration {
private static final String KEYTAB_FILE_DEFAULT = "";
private String servicePrincipal;
private static final String SERVICE_PRINCIPAL_DEFAULT = "";
+ private OutputStream outputStream;
@VisibleForTesting
static final int MAX_MESSAGE_BYTES_DEFAULT = 1024 * 1024;
@@ -49,6 +58,13 @@ public class SubprocessConfiguration {
}
/**
+ * @return the output stream to output messages to.
+ */
+ OutputStream getOutputStream() {
+ return outputStream;
+ }
+
+ /**
* @return the size of the message queue, or the default value if not
* provided
*/
@@ -131,26 +147,56 @@ public class SubprocessConfiguration {
principalOpt.setRequired(false);
options.addOption(principalOpt);
+ final String outputPipeLongOpt = "outputPipe";
+ Option outputPipeOpt = new Option(
+ "o", outputPipeLongOpt, /* hasArg= */ true,
+ "The pipe to output messages to. If not set, outputs to stdout (this " +
+ "is generally unsafe and should only be used in tests)");
+ outputPipeOpt.setRequired(false);
+ options.addOption(outputPipeOpt);
+
CommandLineParser parser = new BasicParser();
+ String outputPipePath;
try {
CommandLine cmd = parser.parse(options, args);
String queueSize = cmd.getOptionValue(queueSizeLongOpt);
this.queueSize = queueSize == null ?
QUEUE_SIZE_DEFAULT : Integer.parseInt(queueSize);
+
String maxParserThreads = cmd.getOptionValue(maxMsgParserThreadsLongOpt);
this.maxMsgParserThreads = maxParserThreads == null ?
MAX_MSG_PARSER_THREADS_DEFAULT : Integer.parseInt(maxParserThreads);
+
String maxMsgBytes = cmd.getOptionValue(maxMsgBytesLongOpt);
this.maxMsgBytes = maxMsgBytes == null ?
MAX_MESSAGE_BYTES_DEFAULT : Integer.parseInt(maxMsgBytes);
+
String keytab = cmd.getOptionValue(keytabFileLongOpt);
this.keytabFile = keytab == null ?
KEYTAB_FILE_DEFAULT : keytab;
+
String principal = cmd.getOptionValue(principalLongOpt);
this.servicePrincipal = principal == null ?
SERVICE_PRINCIPAL_DEFAULT : principal;
+
+ outputPipePath = cmd.getOptionValue(outputPipeLongOpt);
} catch (ParseException e) {
throw new KuduSubprocessException("Cannot parse the subprocess command line", e);
}
+
+ try {
+ if (outputPipePath == null) {
+ this.outputStream = new SubprocessOutputStream(System.out);
+ } else {
+ // If we're not sending messages to System.out, redirect our logs to it.
+ BasicConfigurator.configure();
+ RandomAccessFile outputFile = new RandomAccessFile(new File(outputPipePath), "rw");
+ this.outputStream = new FileOutputStream(outputFile.getFD());
+ }
+ } catch (FileNotFoundException e) {
+ throw new KuduSubprocessException("Output file not found", e);
+ } catch (IOException e) {
+ throw new KuduSubprocessException("IO error opening file", e);
+ }
}
}
diff --git a/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/SubprocessExecutor.java b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/SubprocessExecutor.java
index e9c06e5..1363ea0 100644
--- a/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/SubprocessExecutor.java
+++ b/java/kudu-subprocess/src/main/java/org/apache/kudu/subprocess/SubprocessExecutor.java
@@ -103,8 +103,8 @@ public class SubprocessExecutor {
// a message to ensure the underlying buffer can hold the entire message before
// flushing.
try (BufferedInputStream in = new BufferedInputStream(System.in);
- BufferedOutputStream out = new BufferedOutputStream(
- new SubprocessOutputStream(System.out), maxMessageBytes)) {
+ BufferedOutputStream out = new BufferedOutputStream(conf.getOutputStream(),
+ maxMessageBytes)) {
MessageIO messageIO = new MessageIO(maxMessageBytes, in, out);
// Start a single reader thread and run the task asynchronously.
diff --git a/src/kudu/master/catalog_manager.cc b/src/kudu/master/catalog_manager.cc
index 53e7919..14e4173 100644
--- a/src/kudu/master/catalog_manager.cc
+++ b/src/kudu/master/catalog_manager.cc
@@ -787,7 +787,8 @@ CatalogManager::CatalogManager(Master* master)
if (SentryAuthzProvider::IsEnabled()) {
authz_provider_.reset(new SentryAuthzProvider(master_->metric_entity()));
} else if (RangerAuthzProvider::IsEnabled()) {
- authz_provider_.reset(new RangerAuthzProvider(master_->metric_entity()));
+ authz_provider_.reset(new RangerAuthzProvider(master_->fs_manager()->env(),
+ master_->metric_entity()));
} else {
authz_provider_.reset(new DefaultAuthzProvider);
}
diff --git a/src/kudu/master/ranger_authz_provider.cc b/src/kudu/master/ranger_authz_provider.cc
index f33d8f7..43ecfe5 100644
--- a/src/kudu/master/ranger_authz_provider.cc
+++ b/src/kudu/master/ranger_authz_provider.cc
@@ -31,7 +31,9 @@
DECLARE_string(ranger_config_path);
namespace kudu {
+
class MetricEntity;
+class Env;
namespace master {
@@ -43,8 +45,9 @@ using kudu::ranger::RangerClient;
using std::string;
using std::unordered_set;
-RangerAuthzProvider::RangerAuthzProvider(const scoped_refptr<MetricEntity>& metric_entity) :
- client_(metric_entity) {}
+RangerAuthzProvider::RangerAuthzProvider(Env* env,
+ const scoped_refptr<MetricEntity>& metric_entity) :
+ client_(env, metric_entity) {}
Status RangerAuthzProvider::Start() {
RETURN_NOT_OK(client_.Start());
diff --git a/src/kudu/master/ranger_authz_provider.h b/src/kudu/master/ranger_authz_provider.h
index bdfb11a..6dabf33 100644
--- a/src/kudu/master/ranger_authz_provider.h
+++ b/src/kudu/master/ranger_authz_provider.h
@@ -28,6 +28,7 @@
namespace kudu {
+class Env;
class MetricEntity;
class SchemaPB;
@@ -43,7 +44,7 @@ namespace master {
class RangerAuthzProvider : public AuthzProvider {
public:
- explicit RangerAuthzProvider(const scoped_refptr<MetricEntity>& metric_entity);
+ explicit RangerAuthzProvider(Env* env, const scoped_refptr<MetricEntity>& metric_entity);
Status Start() override;
diff --git a/src/kudu/ranger/ranger_client-test.cc b/src/kudu/ranger/ranger_client-test.cc
index 9a85aaa..b15be50 100644
--- a/src/kudu/ranger/ranger_client-test.cc
+++ b/src/kudu/ranger/ranger_client-test.cc
@@ -33,6 +33,7 @@
#include "kudu/ranger/ranger.pb.h"
#include "kudu/subprocess/server.h"
#include "kudu/subprocess/subprocess.pb.h"
+#include "kudu/util/env.h"
#include "kudu/util/metrics.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
@@ -93,7 +94,7 @@ class MockSubprocessServer : public SubprocessServer {
~MockSubprocessServer() override {}
MockSubprocessServer()
- : SubprocessServer({"mock"}, SubprocessMetrics()) {}
+ : SubprocessServer(Env::Default(), "", {"mock"}, SubprocessMetrics()) {}
Status Execute(SubprocessRequestPB* req,
SubprocessResponsePB* resp) override {
@@ -123,7 +124,7 @@ class MockSubprocessServer : public SubprocessServer {
class RangerClientTest : public KuduTest {
public:
RangerClientTest() :
- client_(METRIC_ENTITY_server.Instantiate(&metric_registry_, "ranger_client-test")) {}
+ client_(env_, METRIC_ENTITY_server.Instantiate(&metric_registry_, "ranger_client-test")) {}
void SetUp() override {
std::unique_ptr<MockSubprocessServer> server(new MockSubprocessServer());
diff --git a/src/kudu/ranger/ranger_client.cc b/src/kudu/ranger/ranger_client.cc
index a2abee1..53371ee 100644
--- a/src/kudu/ranger/ranger_client.cc
+++ b/src/kudu/ranger/ranger_client.cc
@@ -17,8 +17,7 @@
#include "kudu/ranger/ranger_client.h"
-#include <stdint.h>
-
+#include <cstdint>
#include <ostream>
#include <string>
#include <utility>
@@ -34,6 +33,7 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/ranger/ranger.pb.h"
#include "kudu/security/init.h"
+#include "kudu/subprocess/server.h"
#include "kudu/util/env.h"
#include "kudu/util/flag_tags.h"
#include "kudu/util/flag_validators.h"
@@ -59,9 +59,16 @@ TAG_FLAG(ranger_config_path, experimental);
DEFINE_string(ranger_jar_path, "",
"Path to the JAR file containing the Ranger subprocess. "
"If not set, the default JAR file path is expected to be"
- "next to the master binary");
+ "next to the master binary.");
TAG_FLAG(ranger_jar_path, experimental);
+DEFINE_string(ranger_receiver_fifo_dir, "",
+ "Directory in which to create a fifo used to receive messages "
+ "from the Ranger subprocess. Existing fifos at this path will be "
+ "overwritten. If not specified, a fifo will be created in the "
+ "--ranger_config_path directory.");
+TAG_FLAG(ranger_receiver_fifo_dir, advanced);
+
METRIC_DEFINE_histogram(server, ranger_subprocess_execution_time_ms,
"Ranger subprocess execution time (ms)",
kudu::MetricUnit::kMilliseconds,
@@ -155,12 +162,21 @@ static string GetJavaClasspath() {
return Substitute("$0:$1", GetRangerJarPath(), FLAGS_ranger_config_path);
}
-// Builds the arguments for the Ranger subprocess. Specifically pass
-// the principal and keytab file that the Ranger subprocess will log in with
-// if Kerberos is enabled. 'args' has the final arguments.
-// Returns 'OK' if arguments successfully created, error otherwise.
-static Status BuildArgv(vector<string>* argv) {
+
+static string ranger_fifo_base() {
+ DCHECK(!FLAGS_ranger_config_path.empty());
+ const string& fifo_dir = FLAGS_ranger_receiver_fifo_dir.empty() ?
+ FLAGS_ranger_config_path : FLAGS_ranger_receiver_fifo_dir;
+ return JoinPathSegments(fifo_dir, "ranger_receiever_fifo");
+}
+
+// Builds the arguments to start the Ranger subprocess with the given receiver
+// fifo path. Specifically pass the principal and keytab file that the Ranger
+// subprocess will log in with if Kerberos is enabled. 'args' has the final
+// arguments. Returns 'OK' if arguments successfully created, error otherwise.
+static Status BuildArgv(const string& fifo_path, vector<string>* argv) {
DCHECK(argv);
+ DCHECK(!FLAGS_ranger_config_path.empty());
// Pass the required arguments to run the Ranger subprocess.
vector<string> ret = { FLAGS_ranger_java_path, "-cp", GetJavaClasspath(), kMainClass };
// When Kerberos is enabled in Kudu, pass both Kudu principal and keytab file
@@ -174,7 +190,8 @@ static Status BuildArgv(vector<string>* argv) {
ret.emplace_back("-k");
ret.emplace_back(FLAGS_keytab_file);
}
-
+ ret.emplace_back("-o");
+ ret.emplace_back(fifo_path);
*argv = std::move(ret);
return Status::OK();
}
@@ -227,16 +244,17 @@ RangerSubprocessMetrics::RangerSubprocessMetrics(const scoped_refptr<MetricEntit
}
#undef HISTINIT
-RangerClient::RangerClient(const scoped_refptr<MetricEntity>& metric_entity)
- : metric_entity_(metric_entity) {
+RangerClient::RangerClient(Env* env, const scoped_refptr<MetricEntity>& metric_entity)
+ : env_(env), metric_entity_(metric_entity) {
DCHECK(metric_entity);
}
Status RangerClient::Start() {
VLOG(1) << "Initializing Ranger subprocess server";
vector<string> argv;
- RETURN_NOT_OK(BuildArgv(&argv));
- subprocess_.reset(new RangerSubprocess(std::move(argv), metric_entity_));
+ const string fifo_path = subprocess::SubprocessServer::FifoPath(ranger_fifo_base());
+ RETURN_NOT_OK(BuildArgv(fifo_path, &argv));
+ subprocess_.reset(new RangerSubprocess(env_, fifo_path, std::move(argv), metric_entity_));
return subprocess_->Start();
}
diff --git a/src/kudu/ranger/ranger_client.h b/src/kudu/ranger/ranger_client.h
index cdfb8ed..1117785 100644
--- a/src/kudu/ranger/ranger_client.h
+++ b/src/kudu/ranger/ranger_client.h
@@ -32,6 +32,8 @@
namespace kudu {
+class Env;
+
namespace ranger {
struct ActionHash {
@@ -64,7 +66,7 @@ class RangerClient {
};
// Creates a Ranger client.
- explicit RangerClient(const scoped_refptr<MetricEntity>& metric_entity);
+ explicit RangerClient(Env* env, const scoped_refptr<MetricEntity>& metric_entity);
// Starts the RangerClient, initializes the subprocess server.
Status Start() WARN_UNUSED_RESULT;
@@ -105,12 +107,13 @@ class RangerClient {
void ReplaceServerForTests(std::unique_ptr<subprocess::SubprocessServer> server) {
// Creates a dummy RangerSubprocess if it is not initialized.
if (!subprocess_) {
- subprocess_.reset(new RangerSubprocess({""}, metric_entity_));
+ subprocess_.reset(new RangerSubprocess(env_, "", {""}, metric_entity_));
}
subprocess_->ReplaceServerForTests(std::move(server));
}
private:
+ Env* env_;
std::unique_ptr<RangerSubprocess> subprocess_;
scoped_refptr<MetricEntity> metric_entity_;
};
diff --git a/src/kudu/subprocess/server.cc b/src/kudu/subprocess/server.cc
index 4f7b8ae..76f3a3d 100644
--- a/src/kudu/subprocess/server.cc
+++ b/src/kudu/subprocess/server.cc
@@ -17,6 +17,8 @@
#include "kudu/subprocess/server.h"
+#include <unistd.h>
+
#include <csignal>
#include <memory>
#include <ostream>
@@ -33,6 +35,7 @@
#include "kudu/subprocess/subprocess.pb.h"
#include "kudu/subprocess/subprocess_protocol.h"
#include "kudu/util/async_util.h"
+#include "kudu/util/env.h"
#include "kudu/util/flag_tags.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/status.h"
@@ -83,16 +86,25 @@ using strings::Substitute;
namespace kudu {
namespace subprocess {
-SubprocessServer::SubprocessServer(vector<string> subprocess_argv, SubprocessMetrics metrics)
+string SubprocessServer::FifoPath(const string& base) {
+ return Substitute("$0.$1.$2", base, getpid(), Thread::CurrentThreadId());
+}
+
+SubprocessServer::SubprocessServer(Env* env, const string& receiver_file,
+ vector<string> subprocess_argv,
+ SubprocessMetrics metrics)
: call_timeout_(MonoDelta::FromSeconds(FLAGS_subprocess_timeout_secs)),
next_id_(1),
closing_(1),
+ env_(env),
+ receiver_file_(receiver_file),
process_(make_shared<Subprocess>(std::move(subprocess_argv))),
outbound_call_queue_(FLAGS_subprocess_request_queue_size_bytes),
inbound_response_queue_(FLAGS_subprocess_response_queue_size_bytes),
metrics_(std::move(metrics)) {
process_->ShareParentStdin(false);
- process_->ShareParentStdout(false);
+ process_->ShareParentStdout(true);
+ process_->ShareParentStderr(true);
}
SubprocessServer::~SubprocessServer() {
@@ -111,6 +123,7 @@ void SubprocessServer::StartSubprocessThread(const StatusCallback& cb) {
Status SubprocessServer::Init() {
VLOG(2) << "Starting the subprocess";
+
Synchronizer sync;
auto cb = sync.AsStatusCallback();
RETURN_NOT_OK(Thread::Create("subprocess", "start",
@@ -118,11 +131,20 @@ Status SubprocessServer::Init() {
&read_thread_));
RETURN_NOT_OK_PREPEND(sync.Wait(), "Failed to start subprocess");
+ // NOTE: callers should try to ensure each receiver file path is used by a
+ // single subprocess.
+ if (env_->FileExists(receiver_file_)) {
+ RETURN_NOT_OK(env_->DeleteFile(receiver_file_));
+ }
+ // Open the file we'll use for receiving messages.
+ RETURN_NOT_OK(env_->NewFifo(receiver_file_, &receiver_fifo_));
+ RETURN_NOT_OK(receiver_fifo_->OpenForReads());
+
// Start the message protocol.
CHECK(!message_protocol_);
message_protocol_.reset(new SubprocessProtocol(SubprocessProtocol::SerializationMode::PB,
SubprocessProtocol::CloseMode::CLOSE_ON_DESTROY,
- process_->ReleaseChildStdoutFd(),
+ receiver_fifo_->read_fd(),
process_->ReleaseChildStdinFd()));
const int num_threads = FLAGS_subprocess_num_responder_threads;
responder_threads_.resize(num_threads);
@@ -195,6 +217,11 @@ void SubprocessServer::Shutdown() {
for (const auto& t : responder_threads_) {
t->Join();
}
+ // Delete the receiver fifo.
+ receiver_fifo_.reset();
+ if (env_->FileExists(receiver_file_)) {
+ WARN_NOT_OK(env_->DeleteFile(receiver_file_), "Error deleting receiver file");
+ }
// Call any of the remaining callbacks.
std::map<CallId, shared_ptr<SubprocessCall>> calls;
@@ -216,7 +243,6 @@ void SubprocessServer::ReceiveMessagesThread() {
Status s = message_protocol_->ReceiveMessage(&response);
if (s.IsEndOfFile()) {
// The underlying pipe was closed. We're likely shutting down.
- DCHECK_EQ(0, closing_.count());
LOG(INFO) << "Received an EOF from the subprocess";
return;
}
diff --git a/src/kudu/subprocess/server.h b/src/kudu/subprocess/server.h
index ce0a2df..5d46ef7 100644
--- a/src/kudu/subprocess/server.h
+++ b/src/kudu/subprocess/server.h
@@ -47,6 +47,8 @@
namespace kudu {
+class Env;
+class Fifo;
class Subprocess;
class Thread;
@@ -216,7 +218,12 @@ typedef BlockingQueue<ResponsePBAndTimer, ResponseLogicalSize> ResponseQueue;
// Public methods are virtual so a mock server can be used in tests.
class SubprocessServer {
public:
- SubprocessServer(std::vector<std::string> subprocess_argv, SubprocessMetrics metrics);
+ // Returns a path based on 'base' that can be used as a fifo, avoiding
+ // collisions between subprocesses started in different process and threads.
+ static std::string FifoPath(const std::string& base);
+
+ SubprocessServer(Env* env, const std::string& receiver_file,
+ std::vector<std::string> subprocess_argv, SubprocessMetrics metrics);
virtual ~SubprocessServer();
// Initialize the server, starting the subprocess and worker threads.
@@ -270,6 +277,12 @@ class SubprocessServer {
// Latch used to indicate that the server is shutting down.
CountDownLatch closing_;
+ Env* env_;
+ const std::string receiver_file_;
+
+ // The fifo used to receieve messages from the subprocess.
+ std::unique_ptr<Fifo> receiver_fifo_;
+
// The underlying subprocess.
std::shared_ptr<Subprocess> process_;
diff --git a/src/kudu/subprocess/subprocess_proxy-test.cc b/src/kudu/subprocess/subprocess_proxy-test.cc
index ac03cc5..ebfe003 100644
--- a/src/kudu/subprocess/subprocess_proxy-test.cc
+++ b/src/kudu/subprocess/subprocess_proxy-test.cc
@@ -30,6 +30,7 @@
#include "kudu/gutil/casts.h"
#include "kudu/gutil/ref_counted.h"
#include "kudu/gutil/strings/substitute.h"
+#include "kudu/subprocess/server.h"
#include "kudu/subprocess/echo_subprocess.h"
#include "kudu/subprocess/subprocess.pb.h"
#include "kudu/util/env.h"
@@ -64,7 +65,8 @@ class EchoSubprocessTest : public KuduTest {
public:
EchoSubprocessTest()
: metric_entity_(METRIC_ENTITY_server.Instantiate(&metric_registry_,
- "subprocess_proxy-test")) {}
+ "subprocess_proxy-test")),
+ test_dir_(GetTestDataDirectory()) {}
void SetUp() override {
KuduTest::SetUp();
@@ -77,12 +79,15 @@ class EchoSubprocessTest : public KuduTest {
const string bin_dir = DirName(exe);
string java_home;
RETURN_NOT_OK(FindHomeDir("java", bin_dir, &java_home));
+ const string& pipe_file = SubprocessServer::FifoPath(JoinPathSegments(test_dir_, "echo_pipe"));
vector<string> argv = {
Substitute("$0/bin/java", java_home),
"-cp", Substitute("$0/kudu-subprocess.jar", bin_dir),
- "org.apache.kudu.subprocess.echo.EchoSubprocessMain"
+ "org.apache.kudu.subprocess.echo.EchoSubprocessMain",
+ "-o", pipe_file,
};
- echo_subprocess_.reset(new EchoSubprocess(std::move(argv), metric_entity_));
+ echo_subprocess_.reset(new EchoSubprocess(env_, pipe_file, std::move(argv),
+ metric_entity_));
return echo_subprocess_->Start();
}
@@ -90,6 +95,7 @@ class EchoSubprocessTest : public KuduTest {
MetricRegistry metric_registry_;
scoped_refptr<MetricEntity> metric_entity_;
unique_ptr<EchoSubprocess> echo_subprocess_;
+ const string test_dir_;
};
#define GET_HIST(metric_entity, metric_name) \
diff --git a/src/kudu/subprocess/subprocess_proxy.h b/src/kudu/subprocess/subprocess_proxy.h
index d8cf6c7..76289bd 100644
--- a/src/kudu/subprocess/subprocess_proxy.h
+++ b/src/kudu/subprocess/subprocess_proxy.h
@@ -29,6 +29,7 @@
#include "kudu/util/metrics.h"
#include "kudu/util/pb_util.h"
#include "kudu/util/status.h"
+#include "kudu/util/thread.h"
namespace kudu {
namespace subprocess {
@@ -39,8 +40,9 @@ namespace subprocess {
template<class ReqPB, class RespPB, class MetricsPB>
class SubprocessProxy {
public:
- SubprocessProxy(std::vector<std::string> argv, const scoped_refptr<MetricEntity>& entity)
- : server_(new SubprocessServer(std::move(argv), MetricsPB(entity))) {}
+ SubprocessProxy(Env* env, const std::string& receiver_file,
+ std::vector<std::string> argv, const scoped_refptr<MetricEntity>& entity)
+ : server_(new SubprocessServer(env, receiver_file, std::move(argv), MetricsPB(entity))) {}
// Starts the underlying subprocess.
Status Start() {
diff --git a/src/kudu/subprocess/subprocess_server-test.cc b/src/kudu/subprocess/subprocess_server-test.cc
index 90decac..af00559 100644
--- a/src/kudu/subprocess/subprocess_server-test.cc
+++ b/src/kudu/subprocess/subprocess_server-test.cc
@@ -36,6 +36,7 @@
#include "kudu/util/metrics.h"
#include "kudu/util/monotime.h"
#include "kudu/util/path_util.h"
+#include "kudu/util/pb_util.h"
#include "kudu/util/scoped_cleanup.h"
#include "kudu/util/status.h"
#include "kudu/util/stopwatch.h"
@@ -74,6 +75,19 @@ SubprocessRequestPB CreateEchoSubprocessRequestPB(const string& payload,
return request;
}
+Status CheckMessage(const SubprocessResponsePB& resp, const string& expected_msg) {
+ EchoResponsePB echo_resp;
+ if (!resp.response().UnpackTo(&echo_resp)) {
+ return Status::Corruption(Substitute("Failed to unpack echo response: $0",
+ pb_util::SecureDebugString(resp)));
+ }
+ if (expected_msg != echo_resp.data()) {
+ return Status::Corruption(Substitute("Expected: '$0', got: '$1'",
+ expected_msg, echo_resp.data()));
+ }
+ return Status::OK();
+}
+
const char* kHello = "hello world";
} // anonymous namespace
@@ -81,16 +95,17 @@ const char* kHello = "hello world";
class SubprocessServerTest : public KuduTest {
public:
SubprocessServerTest()
- : metric_entity_(METRIC_ENTITY_server.Instantiate(&metric_registry_,
+ : test_dir_(GetTestDataDirectory()),
+ metric_entity_(METRIC_ENTITY_server.Instantiate(&metric_registry_,
"subprocess_server-test")) {}
void SetUp() override {
KuduTest::SetUp();
ASSERT_OK(ResetSubprocessServer());
}
- // Resets the subprocess server to account for any new configuration.
- Status ResetSubprocessServer(int java_queue_size = 0,
- int java_parser_threads = 0) {
+ Status InitSubprocessServer(int java_queue_size,
+ int java_parser_threads,
+ shared_ptr<SubprocessServer>* out) {
// Set up a subprocess server pointing at the kudu-subprocess.jar that
// contains an echo handler and call EchoSubprocessMain.
string exe;
@@ -98,25 +113,34 @@ class SubprocessServerTest : public KuduTest {
const string bin_dir = DirName(exe);
string java_home;
RETURN_NOT_OK(FindHomeDir("java", bin_dir, &java_home));
+ const string pipe_path = SubprocessServer::FifoPath(JoinPathSegments(test_dir_, "echo_pipe"));
vector<string> argv = {
Substitute("$0/bin/java", java_home),
"-cp", Substitute("$0/kudu-subprocess.jar", bin_dir),
- "org.apache.kudu.subprocess.echo.EchoSubprocessMain"
+ "org.apache.kudu.subprocess.echo.EchoSubprocessMain",
+ "-o", pipe_path,
};
if (java_queue_size > 0) {
- argv.emplace_back("q");
+ argv.emplace_back("-q");
argv.emplace_back(std::to_string(java_queue_size));
}
if (java_parser_threads > 0) {
- argv.emplace_back("p");
+ argv.emplace_back("-p");
argv.emplace_back(std::to_string(java_parser_threads));
}
- server_ = make_shared<SubprocessServer>(std::move(argv),
- EchoSubprocessMetrics(metric_entity_));
- return server_->Init();
+ *out = make_shared<SubprocessServer>(env_, pipe_path, std::move(argv),
+ EchoSubprocessMetrics(metric_entity_));
+ return (*out)->Init();
+ }
+
+ // Resets the subprocess server to account for any new configuration.
+ Status ResetSubprocessServer(int java_queue_size = 0,
+ int java_parser_threads = 0) {
+ return InitSubprocessServer(java_queue_size, java_parser_threads, &server_);
}
protected:
+ const string test_dir_;
MetricRegistry metric_registry_;
scoped_refptr<MetricEntity> metric_entity_;
shared_ptr<SubprocessServer> server_;
@@ -170,11 +194,9 @@ TEST_F(SubprocessServerTest, TestManyConcurrentCalls) {
reqs_sent, elapsed_seconds, reqs_sent / elapsed_seconds);
for (int t = 0; t < kNumThreads; t++) {
for (int i = 0; i < kNumPerThread; i++) {
- EchoResponsePB echo_resp;
- ASSERT_TRUE(responses[t][i].response().UnpackTo(&echo_resp));
EchoRequestPB echo_req;
requests[t][i].request().UnpackTo(&echo_req);
- ASSERT_EQ(echo_resp.data(), echo_req.data());
+ ASSERT_OK(CheckMessage(responses[t][i], echo_req.data()));
}
}
}
@@ -280,6 +302,39 @@ TEST_F(SubprocessServerTest, TestInitFromThread) {
ASSERT_OK(server_->Execute(&request, &response));
}
+// Test that we've configured out subprocess server such that we can run it
+// from multiple threads without having them collide with each other.
+TEST_F(SubprocessServerTest, TestRunFromMultipleThreads) {
+ const int kNumThreads = 3;
+ vector<thread> threads;
+ vector<Status> results(kNumThreads);
+#define EXIT_NOT_OK(s, n) do { \
+ Status _s = (s); \
+ if (!_s.ok()) { \
+ results[n] = _s; \
+ return; \
+ } \
+} while (0);
+
+ for (int i = 0; i < kNumThreads; i++) {
+ threads.emplace_back([&, i] {
+ shared_ptr<SubprocessServer> server;
+ EXIT_NOT_OK(InitSubprocessServer(0, 0, &server), i);
+ const string msg = Substitute("$0 bottles of tea on the wall", i);
+ SubprocessRequestPB req = CreateEchoSubprocessRequestPB(msg);
+ SubprocessResponsePB resp;
+ EXIT_NOT_OK(server->Execute(&req, &resp), i);
+ EXIT_NOT_OK(CheckMessage(resp, msg), i);
+ });
+ }
+ for (auto& t : threads) {
+ t.join();
+ }
+ for (const auto& r : results) {
+ ASSERT_OK(r);
+ }
+}
+
} // namespace subprocess
} // namespace kudu
[kudu] 04/04: env: add a fifo class
Posted by aw...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
awong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 5cd26cfad9a5796080d1242fff2090c05eaa1f4a
Author: Andrew Wong <aw...@cloudera.com>
AuthorDate: Sat Mar 28 02:40:10 2020 -0700
env: add a fifo class
This adds a basic Fifo class and a PosixFifo variant that encapsulates
the usage of mkfifo and associated functionality. My immediate need is
just to return the FDs of the opened fifo, so the interface is quite
bare. That said, this will be good enough to move the SubprocessServer
away from using the process stdout FD for IPC.
Change-Id: I23920457d695b84509937694daea53d61f445e27
Reviewed-on: http://gerrit.cloudera.org:8080/15584
Reviewed-by: Adar Dembo <ad...@cloudera.com>
Reviewed-by: Hao Hao <ha...@cloudera.com>
Tested-by: Kudu Jenkins
---
src/kudu/util/env-test.cc | 32 ++++++++++++++++
src/kudu/util/env.h | 27 ++++++++++++-
src/kudu/util/env_posix.cc | 94 ++++++++++++++++++++++++++++++++++++----------
3 files changed, 132 insertions(+), 21 deletions(-)
diff --git a/src/kudu/util/env-test.cc b/src/kudu/util/env-test.cc
index 364a7ba..6dda618 100644
--- a/src/kudu/util/env-test.cc
+++ b/src/kudu/util/env-test.cc
@@ -42,6 +42,7 @@
#include <memory>
#include <ostream>
#include <string>
+#include <thread>
#include <unordered_set>
#include <utility>
#include <vector>
@@ -84,6 +85,7 @@ namespace kudu {
using std::pair;
using std::shared_ptr;
using std::string;
+using std::thread;
using std::unique_ptr;
using std::unordered_set;
using std::vector;
@@ -1210,5 +1212,35 @@ TEST_F(TestEnv, TestCreateSymlink) {
ASSERT_TRUE(env_->FileExists(JoinPathSegments(kDst, "foobar")));
}
+TEST_F(TestEnv, TestCreateFifo) {
+ const string kFifo = JoinPathSegments(test_dir_, "fifo");
+ unique_ptr<Fifo> fifo;
+ // Open a fifo for reads and writes.
+ ASSERT_OK(env_->NewFifo(kFifo, &fifo));
+ thread t([&] {
+ ASSERT_OK(fifo->OpenForReads());
+ });
+ ASSERT_OK(fifo->OpenForWrites());
+ t.join();
+
+ Slice data("it's the final countdown");
+ ssize_t written;
+ RETRY_ON_EINTR(written, write(fifo->write_fd(), data.data(), data.size()));
+ ASSERT_EQ(data.size(), written);
+ char buf[32];
+ ssize_t n;
+ RETRY_ON_EINTR(n, read(fifo->read_fd(), buf, arraysize(buf)));
+ ASSERT_EQ(data.size(), n);
+ Slice read_data(buf, n);
+ ASSERT_EQ(data, read_data);
+
+ // Trying to create the same fifo should fail.
+ Status s = env_->NewFifo(kFifo, &fifo);
+ ASSERT_TRUE(s.IsAlreadyPresent()) << s.ToString();
+
+ // Until our fifo gets deleted.
+ ASSERT_OK(env_->DeleteFile(kFifo));
+ ASSERT_OK(env_->NewFifo(kFifo, &fifo));
+}
} // namespace kudu
diff --git a/src/kudu/util/env.h b/src/kudu/util/env.h
index 8d818e9..9a2d7bc 100644
--- a/src/kudu/util/env.h
+++ b/src/kudu/util/env.h
@@ -28,6 +28,7 @@
namespace kudu {
class faststring;
+class Fifo;
class FileLock;
class RandomAccessFile;
class RWFile;
@@ -142,12 +143,16 @@ class Env {
const std::string& fname,
std::unique_ptr<RWFile>* result) = 0;
- // Same as abovoe for NewTempWritableFile(), but for an RWFile.
+ // Same as above for NewTempWritableFile(), but for an RWFile.
virtual Status NewTempRWFile(const RWFileOptions& opts,
const std::string& name_template,
std::string* created_filename,
std::unique_ptr<RWFile>* res) = 0;
+ // Creates a new fifo.
+ virtual Status NewFifo(const std::string& fname,
+ std::unique_ptr<Fifo>* fifo) = 0;
+
// Returns true iff the named file exists.
virtual bool FileExists(const std::string& fname) = 0;
@@ -383,6 +388,26 @@ class File {
virtual const std::string& filename() const = 0;
};
+// A simple fifo abstraction.
+class Fifo : public File {
+ public:
+ // Opens the fifo for reads. This will wait until the fifo has also been
+ // opened for writes.
+ virtual Status OpenForReads() = 0;
+
+ // Opens the fifo for writes. This will wait until the fifo has also been
+ // opened for reads.
+ virtual Status OpenForWrites() = 0;
+
+ // Returns the read fd, set when opened for reads. The fifo must have been
+ // opened for reads before calling.
+ virtual int read_fd() const = 0;
+
+ // Returns the write fd, set when opened for writes. The fifo must have been
+ // opened for writes before calling.
+ virtual int write_fd() const = 0;
+};
+
// A file abstraction for reading sequentially through a file
class SequentialFile : public File {
public:
diff --git a/src/kudu/util/env_posix.cc b/src/kudu/util/env_posix.cc
index 1fd4507..3590fd1 100644
--- a/src/kudu/util/env_posix.cc
+++ b/src/kudu/util/env_posix.cc
@@ -286,6 +286,13 @@ ssize_t pwritev(int fd, const struct iovec* iovec, int count, off_t offset) {
}
#endif
+void DoClose(int fd) {
+ int err;
+ RETRY_ON_EINTR(err, close(fd));
+ if (PREDICT_FALSE(err != 0)) {
+ PLOG(WARNING) << "Failed to close fd " << fd;
+ }
+}
// Close file descriptor when object goes out of scope.
class ScopedFdCloser {
@@ -296,11 +303,7 @@ class ScopedFdCloser {
~ScopedFdCloser() {
ThreadRestrictions::AssertIOAllowed();
- int err;
- RETRY_ON_EINTR(err, ::close(fd_));
- if (PREDICT_FALSE(err != 0)) {
- PLOG(WARNING) << "Failed to close fd " << fd_;
- }
+ DoClose(fd_);
}
private:
@@ -348,6 +351,16 @@ Status DoSync(int fd, const string& filename) {
return Status::OK();
}
+Status DoOpen(const string& filename, int flags, const string& reason, int* fd) {
+ int f;
+ RETRY_ON_EINTR(f, open(filename.c_str(), flags));
+ if (f == -1) {
+ return IOError(Substitute("Error opening for $0: $1", reason, filename), errno);
+ }
+ *fd = f;
+ return Status::OK();
+}
+
Status DoOpen(const string& filename, Env::OpenMode mode, int* fd) {
MAYBE_RETURN_EIO(filename, IOError(Env::kInjectedFailureStatusMsg, EIO));
ThreadRestrictions::AssertIOAllowed();
@@ -556,6 +569,49 @@ const char* ResourceLimitTypeToMacosRlimit(Env::ResourceLimitType t) {
}
#endif
+class PosixFifo : public Fifo {
+ public:
+ explicit PosixFifo(string fname) : filename_(std::move(fname)) {}
+
+ const string& filename() const override {
+ return filename_;
+ }
+
+ Status OpenForReads() override {
+ CHECK_EQ(-1, read_fd_);
+ return DoOpen(filename_, O_RDONLY, "reads", &read_fd_);
+ }
+
+ Status OpenForWrites() override {
+ CHECK_EQ(-1, write_fd_);
+ return DoOpen(filename_, O_WRONLY, "writes", &write_fd_);
+ }
+
+ int read_fd() const override {
+ CHECK_NE(-1, read_fd_);
+ return read_fd_;
+ }
+
+ int write_fd() const override {
+ CHECK_NE(-1, write_fd_);
+ return write_fd_;
+ }
+
+ ~PosixFifo() {
+ if (read_fd_ != -1) {
+ DoClose(read_fd_);
+ }
+ if (write_fd_ != -1) {
+ DoClose(write_fd_);
+ }
+ }
+
+ private:
+ const string filename_;
+ int read_fd_ = -1;
+ int write_fd_ = -1;
+};
+
class PosixSequentialFile: public SequentialFile {
private:
const string filename_;
@@ -614,11 +670,7 @@ class PosixRandomAccessFile: public RandomAccessFile {
PosixRandomAccessFile(string fname, int fd)
: filename_(std::move(fname)), fd_(fd) {}
~PosixRandomAccessFile() {
- int err;
- RETRY_ON_EINTR(err, close(fd_));
- if (PREDICT_FALSE(err != 0)) {
- PLOG(WARNING) << "Failed to close " << filename_;
- }
+ DoClose(fd_);
}
virtual Status Read(uint64_t offset, Slice result) const OVERRIDE {
@@ -1171,6 +1223,16 @@ class PosixEnv : public Env {
return Status::OK();
}
+ virtual Status NewFifo(const string& fname, unique_ptr<Fifo>* fifo) override {
+ TRACE_EVENT1("io", "PosixEnv::NewFifo", "path", fname);
+ int m = mkfifo(fname.c_str(), 0666);
+ if (m != 0) {
+ return IOError(Substitute("Error creating fifo $0", fname), errno);
+ }
+ fifo->reset(new PosixFifo(fname));
+ return Status::OK();
+ }
+
virtual bool FileExists(const string& fname) OVERRIDE {
TRACE_EVENT1("io", "PosixEnv::FileExists", "path", fname);
ThreadRestrictions::AssertIOAllowed();
@@ -1397,11 +1459,7 @@ class PosixEnv : public Env {
result = IOError(fname, errno);
} else if (LockOrUnlock(fd, true) == -1) {
result = IOError("lock " + fname, errno);
- int err;
- RETRY_ON_EINTR(err, close(fd));
- if (PREDICT_FALSE(err != 0)) {
- PLOG(WARNING) << "Failed to close fd " << fd;
- }
+ DoClose(fd);
} else {
auto my_lock = new PosixFileLock;
my_lock->fd_ = fd;
@@ -1418,11 +1476,7 @@ class PosixEnv : public Env {
if (LockOrUnlock(my_lock->fd_, false) == -1) {
result = IOError("unlock", errno);
}
- int err;
- RETRY_ON_EINTR(err, close(my_lock->fd_));
- if (PREDICT_FALSE(err != 0)) {
- PLOG(WARNING) << "Failed to close fd " << my_lock->fd_;
- }
+ DoClose(my_lock->fd_);
return result;
}