You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sentry.apache.org by va...@apache.org on 2017/01/07 04:48:20 UTC
sentry git commit: SENTRY-1583: Refactor ZK/Curator code (Alexander
Kolbasov via Vamsee Yarlagadda, Reviewed by: Misha Dmitriev,
Vamsee Yarlagadda)
Repository: sentry
Updated Branches:
refs/heads/sentry-ha-redesign cf8a16f2a -> 6125ac9bd
SENTRY-1583: Refactor ZK/Curator code (Alexander Kolbasov via Vamsee Yarlagadda, Reviewed by: Misha Dmitriev, Vamsee Yarlagadda)
Project: http://git-wip-us.apache.org/repos/asf/sentry/repo
Commit: http://git-wip-us.apache.org/repos/asf/sentry/commit/6125ac9b
Tree: http://git-wip-us.apache.org/repos/asf/sentry/tree/6125ac9b
Diff: http://git-wip-us.apache.org/repos/asf/sentry/diff/6125ac9b
Branch: refs/heads/sentry-ha-redesign
Commit: 6125ac9bd541cb141beafa844fc7e32157c532d0
Parents: cf8a16f
Author: Vamsee Yarlagadda <va...@cloudera.com>
Authored: Fri Jan 6 20:44:34 2017 -0800
Committer: Vamsee Yarlagadda <va...@cloudera.com>
Committed: Fri Jan 6 20:44:43 2017 -0800
----------------------------------------------------------------------
pom.xml | 2 +-
.../core/common/utils/SentryConstants.java | 1 -
.../db/service/persistent/HAContext.java | 270 +++++++++++++++++++
.../db/service/thrift/SentryMetrics.java | 2 +-
.../apache/sentry/service/thrift/Activator.java | 151 -----------
.../sentry/service/thrift/Activators.java | 75 ------
.../sentry/service/thrift/LeaderStatus.java | 181 -------------
.../service/thrift/LeaderStatusAdaptor.java | 207 --------------
.../service/thrift/LeaderStatusMonitor.java | 270 +++++++++++++++++++
.../sentry/service/thrift/SentryService.java | 85 +++---
.../sentry/service/thrift/ServiceConstants.java | 6 +-
.../db/service/thrift/TestActivator.java | 8 -
.../thrift/TestSentryServiceMetrics.java | 1 -
.../sentry/service/thrift/TestLeaderStatus.java | 245 -----------------
14 files changed, 587 insertions(+), 917 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index df26edf..b9282e7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -62,7 +62,7 @@ limitations under the License.
<commons-pool2.version>2.2</commons-pool2.version>
<commons.lang.version>2.6</commons.lang.version>
<commons.logging.version>1.2</commons.logging.version>
- <curator.version>2.7.1</curator.version>
+ <curator.version>2.11.1</curator.version>
<datanucleus.maven.plugin.version>4.0.1</datanucleus.maven.plugin.version>
<derby.version>10.10.2.0</derby.version>
<easymock.version>3.0</easymock.version>
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-core/sentry-core-common/src/main/java/org/apache/sentry/core/common/utils/SentryConstants.java
----------------------------------------------------------------------
diff --git a/sentry-core/sentry-core-common/src/main/java/org/apache/sentry/core/common/utils/SentryConstants.java b/sentry-core/sentry-core-common/src/main/java/org/apache/sentry/core/common/utils/SentryConstants.java
index c094058..4ed1361 100644
--- a/sentry-core/sentry-core-common/src/main/java/org/apache/sentry/core/common/utils/SentryConstants.java
+++ b/sentry-core/sentry-core-common/src/main/java/org/apache/sentry/core/common/utils/SentryConstants.java
@@ -42,5 +42,4 @@ public class SentryConstants {
public static final String ACCESS_ALLOW_URI_PER_DB_POLICYFILE = "sentry.allow.uri.db.policyfile";
public static final String SENTRY_ZK_JAAS_NAME = "Sentry";
- public static final String CURRENT_INCARNATION_ID_KEY = "current.incarnation.key";
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/HAContext.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/HAContext.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/HAContext.java
new file mode 100644
index 0000000..00eec4e
--- /dev/null
+++ b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/persistent/HAContext.java
@@ -0,0 +1,270 @@
+/*
+ * 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.
+ */
+
+package org.apache.sentry.provider.db.service.persistent;
+
+import com.google.common.base.Strings;
+import com.google.common.collect.Lists;
+import org.apache.curator.RetryPolicy;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
+import org.apache.curator.framework.api.ACLProvider;
+import org.apache.curator.framework.imps.CuratorFrameworkState;
+import org.apache.curator.framework.imps.DefaultACLProvider;
+import org.apache.curator.framework.recipes.leader.LeaderSelector;
+import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.SecurityUtil;
+import org.apache.sentry.service.thrift.JaasConfiguration;
+import org.apache.zookeeper.ZooDefs.Perms;
+import org.apache.zookeeper.client.ZooKeeperSaslClient;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.List;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.sentry.service.thrift.ServiceConstants.ServerConfig.*;
+
+/**
+ * HAContext stores the global ZooKeeper related context.
+ * <p>
+ * This class is a singleton - only one ZooKeeper context is maintained.
+ */
+public final class HAContext implements AutoCloseable {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(HAContext.class);
+ private static HAContext serverHAContext = null;
+ private static boolean aclUnChecked = true;
+
+ private static final String SENTRY_ZK_JAAS_NAME = "SentryClient";
+ private final String zookeeperQuorum;
+ private final String namespace;
+
+ private final boolean zkSecure;
+ private final List<ACL> saslACL;
+
+ private final CuratorFramework curatorFramework;
+
+ private HAContext(Configuration conf) throws IOException {
+ this.zookeeperQuorum = conf.get(SENTRY_HA_ZOOKEEPER_QUORUM, "");
+ int retriesMaxCount = conf.getInt(SENTRY_HA_ZOOKEEPER_RETRIES_MAX_COUNT,
+ SENTRY_HA_ZOOKEEPER_RETRIES_MAX_COUNT_DEFAULT);
+ int sleepMsBetweenRetries = conf.getInt(SENTRY_HA_ZOOKEEPER_SLEEP_BETWEEN_RETRIES_MS,
+ SENTRY_HA_ZOOKEEPER_SLEEP_BETWEEN_RETRIES_MS_DEFAULT);
+ this.namespace = conf.get(SENTRY_HA_ZOOKEEPER_NAMESPACE,
+ SENTRY_HA_ZOOKEEPER_NAMESPACE_DEFAULT);
+ this.zkSecure = conf.getBoolean(SENTRY_HA_ZOOKEEPER_SECURITY,
+ SENTRY_HA_ZOOKEEPER_SECURITY_DEFAULT);
+ this.validateConf();
+ ACLProvider aclProvider;
+ if (zkSecure) {
+ LOGGER.info("Connecting to ZooKeeper with SASL/Kerberos and using 'sasl' ACLs");
+ this.setJaasConfiguration(conf);
+ System.setProperty(ZooKeeperSaslClient.LOGIN_CONTEXT_NAME_KEY,
+ SENTRY_ZK_JAAS_NAME);
+ saslACL = Lists.newArrayList();
+ saslACL.add(new ACL(Perms.ALL, new Id("sasl", getServicePrincipal(conf,
+ PRINCIPAL))));
+ saslACL.add(new ACL(Perms.ALL, new Id("sasl", getServicePrincipal(conf,
+ SERVER_HA_ZOOKEEPER_CLIENT_PRINCIPAL))));
+ aclProvider = new SASLOwnerACLProvider();
+ String allowConnect = conf.get(ALLOW_CONNECT);
+
+ if (!Strings.isNullOrEmpty(allowConnect)) {
+ for (String principal : allowConnect.split("\\s*,\\s*")) {
+ LOGGER.info("Adding acls for " + principal);
+ saslACL.add(new ACL(Perms.ALL, new Id("sasl", principal)));
+ }
+ }
+ } else {
+ saslACL = null;
+ LOGGER.info("Connecting to ZooKeeper without authentication");
+ aclProvider = new DefaultACLProvider();
+ }
+
+ RetryPolicy retryPolicy = new ExponentialBackoffRetry(sleepMsBetweenRetries, retriesMaxCount);
+ this.curatorFramework = CuratorFrameworkFactory.builder()
+ .namespace(this.namespace)
+ .connectString(this.zookeeperQuorum)
+ .retryPolicy(retryPolicy)
+ .aclProvider(aclProvider)
+ .build();
+ }
+
+ private void start() {
+ if (curatorFramework.getState() != CuratorFrameworkState.STARTED) {
+ curatorFramework.start();
+ }
+ }
+
+ /**
+ * Create a singleton instance of ZooKeeper context (if needed) and return it.
+ * The instance returned is already running.
+ *
+ * @param conf Configuration, The following keys are used:
+ * <ul>
+ * <li>SENTRY_HA_ZOOKEEPER_QUORUM</li>
+ * <li>SENTRY_HA_ZOOKEEPER_RETRIES_MAX_COUNT</li>
+ * <li>SENTRY_HA_ZOOKEEPER_SLEEP_BETWEEN_RETRIES_MS</li>
+ * <li>SENTRY_HA_ZOOKEEPER_NAMESPACE</li>
+ * <li>SENTRY_HA_ZOOKEEPER_SECURITY</li>
+ * <li>LOGIN_CONTEXT_NAME_KEY</li>
+ * <li>PRINCIPAL</li>
+ * <li>SERVER_HA_ZOOKEEPER_CLIENT_PRINCIPAL</li>
+ * <li>ALLOW_CONNECT</li>
+ * <li>SERVER_HA_ZOOKEEPER_CLIENT_TICKET_CACHE</li>
+ * <li>SERVER_HA_ZOOKEEPER_CLIENT_KEYTAB</li>
+ * <li>RPC_ADDRESS</li>
+ * </ul>
+ * @return Global ZooKeeper context.
+ * @throws Exception
+ */
+ static synchronized HAContext getHAContext(Configuration conf) throws IOException {
+ if (serverHAContext != null) {
+ return serverHAContext;
+ }
+ serverHAContext = new HAContext(conf);
+
+ serverHAContext.start();
+ Runtime.getRuntime().addShutdownHook(new Thread() {
+ @Override
+ public void run() {
+ LOGGER.info("ShutdownHook closing curator framework");
+ try {
+ serverHAContext.close();
+ } catch (Throwable t) {
+ LOGGER.error("Error stopping curator framework", t);
+ }
+ }
+ });
+ return serverHAContext;
+ }
+
+ /**
+ * HA context for server which verifies the ZK ACLs on namespace
+ *
+ * @param conf Configuration - see {@link #getHAContext(Configuration)}
+ * @return Server ZK context
+ * @throws Exception
+ */
+ public static HAContext getHAServerContext(Configuration conf) throws Exception {
+ HAContext serverContext = getHAContext(conf);
+ serverContext.checkAndSetACLs();
+ return serverContext;
+ }
+
+ private void validateConf() {
+ checkNotNull(zookeeperQuorum, "Zookeeper Quorum should not be null.");
+ checkNotNull(namespace, "Zookeeper namespace should not be null.");
+ }
+
+ private static String getServicePrincipal(Configuration conf, String confProperty)
+ throws IOException {
+ String principal = checkNotNull(conf.get(confProperty));
+ checkArgument(!principal.isEmpty(), "Server principal is empty.");
+ return principal.split("[/@]")[0];
+ }
+
+ private void checkAndSetACLs() throws Exception {
+ if (zkSecure && aclUnChecked) {
+ // If znodes were previously created without security enabled, and now it is, we need to go
+ // through all existing znodes and set the ACLs for them. This is done just once at the startup
+ // We can't get the namespace znode through curator; have to go through zk client
+ String newNamespace = "/" + curatorFramework.getNamespace();
+ if (curatorFramework.getZookeeperClient().getZooKeeper().exists(newNamespace, null) != null) {
+ List<ACL> acls = curatorFramework.getZookeeperClient().getZooKeeper().getACL(newNamespace, new Stat());
+ if (acls.isEmpty() || !acls.get(0).getId().getScheme().equals("sasl")) {
+ LOGGER.info("'sasl' ACLs not set; setting...");
+ List<String> children = curatorFramework.getZookeeperClient().getZooKeeper().getChildren(newNamespace,
+ null);
+ for (String child : children) {
+ this.checkAndSetACLs("/" + child);
+ }
+ curatorFramework.getZookeeperClient().getZooKeeper().setACL(newNamespace, saslACL, -1);
+ }
+ }
+ aclUnChecked = false;
+ }
+ }
+
+ private void checkAndSetACLs(String path) throws Exception {
+ LOGGER.info("Setting acls on " + path);
+ List<String> children = curatorFramework.getChildren().forPath(path);
+ for (String child : children) {
+ this.checkAndSetACLs(path + "/" + child);
+ }
+ curatorFramework.setACL().withACL(saslACL).forPath(path);
+ }
+
+ // This gets ignored during most tests, see ZKXTestCaseWithSecurity#setupZKServer()
+ private void setJaasConfiguration(Configuration conf) throws IOException {
+ if ("false".equalsIgnoreCase(conf.get(
+ SERVER_HA_ZOOKEEPER_CLIENT_TICKET_CACHE,
+ SERVER_HA_ZOOKEEPER_CLIENT_TICKET_CACHE_DEFAULT))) {
+ String keytabFile = conf.get(SERVER_HA_ZOOKEEPER_CLIENT_KEYTAB);
+ checkArgument(!keytabFile.isEmpty(), "Keytab File is empty.");
+ String principal = conf.get(SERVER_HA_ZOOKEEPER_CLIENT_PRINCIPAL);
+ principal = SecurityUtil.getServerPrincipal(principal,
+ conf.get(RPC_ADDRESS, RPC_ADDRESS_DEFAULT));
+ checkArgument(!principal.isEmpty(), "Kerberos principal is empty.");
+
+ // This is equivalent to writing a jaas.conf file and setting the system property,
+ // "java.security.auth.login.config", to point to it (but this way we don't have to write
+ // a file, and it works better for the tests)
+ JaasConfiguration.addEntryForKeytab(SENTRY_ZK_JAAS_NAME, principal, keytabFile);
+ } else {
+ // Create jaas conf for ticket cache
+ JaasConfiguration.addEntryForTicketCache(SENTRY_ZK_JAAS_NAME);
+ }
+ javax.security.auth.login.Configuration.setConfiguration(JaasConfiguration.getInstance());
+ }
+
+ /**
+ * Create a new Curator leader szselector
+ * @param path Zookeeper path
+ * @param listener Curator listener for leader selection changes
+ * @return an instance of leader selector associated with the running curator framework
+ */
+ public LeaderSelector newLeaderSelector(String path, LeaderSelectorListener listener) {
+ return new LeaderSelector(this.curatorFramework, path, listener);
+ }
+
+ @Override
+ public void close() throws Exception {
+ this.curatorFramework.close();
+ }
+
+ private class SASLOwnerACLProvider implements ACLProvider {
+ @Override
+ public List<ACL> getDefaultAcl() {
+ return saslACL;
+ }
+
+ @Override
+ public List<ACL> getAclForPath(String path) {
+ return saslACL;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/thrift/SentryMetrics.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/thrift/SentryMetrics.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/thrift/SentryMetrics.java
index f80605c..39539b9 100644
--- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/thrift/SentryMetrics.java
+++ b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/provider/db/service/thrift/SentryMetrics.java
@@ -134,7 +134,7 @@ public final class SentryMetrics {
public void addSentryServiceGauges(SentryService sentryservice) {
if(!sentryServiceGaugesAdded) {
addGauge(SentryService.class, "is_active", sentryservice.getIsActiveGauge());
- addGauge(SentryService.class, "is_ha", sentryservice.getIsHAGauge());
+ addGauge(SentryService.class, "activated", sentryservice.getBecomeActiveCount());
sentryServiceGaugesAdded = true;
}
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/Activator.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/Activator.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/Activator.java
deleted file mode 100644
index c3df4d8..0000000
--- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/Activator.java
+++ /dev/null
@@ -1,151 +0,0 @@
-/**
- * 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.
- */
-
-package org.apache.sentry.service.thrift;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.Properties;
-
-import com.google.common.annotations.VisibleForTesting;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.sentry.core.common.exception.SentryStandbyException;
-import org.apache.sentry.provider.db.service.persistent.Fencer;
-import org.apache.sentry.provider.db.service.persistent.SentryStore;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.jdo.JDOHelper;
-import javax.jdo.PersistenceManager;
-import javax.jdo.PersistenceManagerFactory;
-
-/**
- * The activator is used to access and modify the activation state of the sentry daemon.
- * In active / active mode, only one daemon can fetch snapshots from HMS and write to the
- * the backend DB, thus we can use the Activator to mark it.
- */
-public class Activator implements Closeable {
- private static final Logger LOGGER = LoggerFactory.getLogger(Activator.class);
-
- /**
- * The DataNucleus PersistenceManagerFactory to use.
- */
- private final PersistenceManagerFactory pmf;
-
- /**
- * The handler for LeaderStatus callbacks.
- */
- private final TransitionHandler handler;
-
- /**
- * LeaderStatus generates callbacks to let us know when we are active or
- * standby. When HA is enabled, it manages ZK sessions.
- */
- private final LeaderStatus leaderStatus;
-
- /**
- * The fencer object.
- */
- private final Fencer fencer;
-
- /**
- * True if the Activator is active.
- *
- * This variable can be read without taking the lock, but must not be
- * written unless we hold the Activator lock.
- */
- private volatile boolean active = false;
-
- public Activator(Configuration conf) throws Exception {
- Properties props = SentryStore.getDataNucleusProperties(conf);
- this.pmf = JDOHelper.getPersistenceManagerFactory(props);
- this.handler = new TransitionHandler();
- this.leaderStatus = new LeaderStatus(handler, conf);
- this.fencer = new Fencer(this.leaderStatus.getIncarnationId(), pmf);
- this.leaderStatus.start();
- }
-
- @Override
- public void close() throws IOException {
- this.leaderStatus.close();
- this.pmf.close();
- }
-
- /**
- * Deactivates this activator.
- */
- @VisibleForTesting
- public void deactivate() throws IOException {
- leaderStatus.becomeStandby();
- }
-
- private class TransitionHandler implements LeaderStatus.Listener {
- @Override
- public void becomeActive() throws Exception {
- synchronized (Activator.this) {
- if (!active) {
- LOGGER.info("Activating " + leaderStatus.getIncarnationId());
- fencer.fence(pmf);
- active = true;
- }
- }
- }
-
- @Override
- public void becomeStandby() {
- synchronized (Activator.this) {
- if (active) {
- LOGGER.info("Deactivating " + leaderStatus.getIncarnationId());
- active = false;
- }
- }
- }
- }
-
- /**
- * Returns true if this Activator considers itself active.
- * Note that you must still use checkSqlFencing or another
- * means of fencing when performing modification operations.
- */
- public boolean isActive() {
- return active;
- }
-
- public synchronized String getIncarnationId() {
- return leaderStatus.getIncarnationId();
- }
-
- public Fencer getFencer() {
- return fencer;
- }
-
- /**
- * Verify that the current SQL transaction is safe.
- */
- public void checkSqlFencing(PersistenceManager pm)
- throws SentryStandbyException {
- // Before invoking the fencer, first check if we believe that we are
- // active. This avoids wasted effort.
- if (!active) {
- throw new SentryStandbyException("The daemon is not active");
- }
- // If we believe that we are active, add a query to the current transaction
- // which will confirm that fact.
- fencer.checkSqlFencing(pm);
- }
-}
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/Activators.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/Activators.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/Activators.java
deleted file mode 100644
index 2926eeb..0000000
--- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/Activators.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/**
- * 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.
- */
-
-package org.apache.sentry.service.thrift;
-
-import java.util.HashMap;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.sentry.core.common.utils.SentryConstants;
-
-/**
- * A global map from incarnation IDs to Activator objects.<p/>
- *
- * This is used to access the current global Activator. Normally there will
- * only be one Activator used in a sentry process. There may be multiple
- * Activator objects in the case where we are running unit tests.
- */
-public enum Activators {
- INSTANCE;
-
- private final HashMap<String, Activator> acts = new HashMap<String, Activator>();
-
- Activators() {}
-
- public synchronized void put(Activator act) {
- acts.put(act.getIncarnationId(), act);
- }
-
- public Activator create(Configuration conf) throws Exception {
- Activator act = new Activator(conf);
- put(act);
- return act;
- }
-
- public Activator get(Configuration conf) {
- String key = conf.get(SentryConstants.CURRENT_INCARNATION_ID_KEY);
- if (key == null) {
- throw new RuntimeException("No " +
- SentryConstants.CURRENT_INCARNATION_ID_KEY + "set.");
- }
- return get(key);
- }
-
- private synchronized Activator get(String incarnationId) {
- Activator act = acts.get(incarnationId);
- if (act == null) {
- throw new RuntimeException("No activator found with " +
- "incarnationId " + incarnationId);
- }
- return act;
- }
-
- public synchronized void remove(Activator act) {
- Activator removed = acts.remove(act.getIncarnationId());
- if (removed == null) {
- throw new RuntimeException("No activator found with " +
- "incarnationId " + act.getIncarnationId());
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/LeaderStatus.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/LeaderStatus.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/LeaderStatus.java
deleted file mode 100644
index 2467921..0000000
--- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/LeaderStatus.java
+++ /dev/null
@@ -1,181 +0,0 @@
-/*
- * 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.
- */
-package org.apache.sentry.service.thrift;
-
-import com.google.common.annotations.VisibleForTesting;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.datanucleus.util.Base64;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.security.SecureRandom;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import static org.apache.sentry.service.thrift.ServiceConstants.ServerConfig.SENTRY_HA_ENABLED;
-import static org.apache.sentry.service.thrift.ServiceConstants.ServerConfig.SENTRY_HA_ENABLED_DEFAULT;
-
-/**
- * Determines the leadership status of the Sentry daemon.
- * It handles both highly-available and non-highly-available configurations.
- */
-final class LeaderStatus implements Closeable {
- private static final Log LOG = LogFactory.getLog(LeaderStatus.class);
-
- /**
- * Callback functions which are invoked when the leader status changes.
- */
- interface Listener {
- /**
- * Attempt to become the leader.
- *
- * @throws Exception On error. If an exception is thrown, we will
- * relinquish leadership.
- */
- void becomeActive() throws Exception;
-
- /**
- * Become the standby. All exceptions thrown from this function will
- * be ignored.
- */
- void becomeStandby();
- }
-
- /**
- * The listener to invoke when our leadership status changes.
- */
- private final Listener listener;
-
- /**
- * The unique ID of this potential leader.
- */
- private final String incarnationId;
-
- /**
- * The LeaderStatusThread, or null if HA is disabled.
- */
- private final LeaderStatusAdaptor leaderStatusAdaptor;
-
- /**
- * True if this object has been closed.<p/>
- *
- * This is an AtomicBoolean so that multiple calls to close only result in one
- * close action.
- */
- private final AtomicBoolean closed = new AtomicBoolean(false);
-
- /**
- * Generate a very long random ID.
- *
- * We want a name that doesn't start with a number, and which
- * contains only letters and numbers. This is important because
- * the incarnation ID gets used in SQL databases to name a table.
- */
- static String generateIncarnationId() {
- SecureRandom srand = new SecureRandom();
- // Why 12? Base64 encodes 12 bytes to a 16 length char array(12 * 8 /6).
- // We need a encoded string of length <= 17, as the length of a
- // fencing table which is prefixed with SENTRY_FENCE_(13 chars)
- // cannot be greater than 30 chars(Oracle limitation)
- byte[] buf = new byte[12];
- srand.nextBytes(buf);
- char[] cbuf = Base64.encode(buf);
- StringBuilder bld = new StringBuilder();
- for (int i = 0; i < cbuf.length; i++) {
- boolean safe;
- if (i == 0) {
- // Some databases can't handle identiifers that start with numbers,
- // so always start with a letter. Also replace '+' or '/' with
- // something safe.
- safe = Character.isLetter(cbuf[i]);
- } else {
- // Replace '+' or '/' with something safe.
- safe = Character.isLetterOrDigit(cbuf[i]);
- }
- if (!safe) {
- bld.append((char)('a' + srand.nextInt(26)));
- } else {
- bld.append(cbuf[i]);
- }
- }
- return bld.toString();
- }
-
- LeaderStatus(Listener listener, Configuration conf) throws Exception {
- this.listener = listener;
- this.incarnationId = generateIncarnationId();
- boolean isHa = conf.
- getBoolean(SENTRY_HA_ENABLED, SENTRY_HA_ENABLED_DEFAULT);
- if (isHa) {
- this.leaderStatusAdaptor = new LeaderStatusAdaptor(incarnationId, conf, listener);
- } else {
- LOG.info("LeaderStatus(incarnationId=" + incarnationId +
- "): HA is disabled.");
- this.leaderStatusAdaptor = null;
- }
- }
-
- public String getIncarnationId() {
- return incarnationId;
- }
-
- public void start() throws Exception {
- if (this.leaderStatusAdaptor != null) {
- this.leaderStatusAdaptor.start();
- } else {
- this.listener.becomeActive();
- }
- }
-
- /**
- * Called by tests to force deactivate(standby) a daemon,
- * so that another daemon becomes active.
- * @throws IOException
- */
- @VisibleForTesting
- public void becomeStandby() throws IOException {
- if (leaderStatusAdaptor != null) {
- leaderStatusAdaptor.deactivate();
- } else {
- try {
- listener.becomeStandby();
- } catch (Throwable t) {
- LOG.error("becomeStandby: " + incarnationId +
- " threw an unexpected exception", t);
- }
- }
- }
-
- @Override
- public void close() throws IOException {
- if (closed.compareAndSet(false, true)) {
- if (leaderStatusAdaptor != null) {
- // Shut down in the HA case.
- leaderStatusAdaptor.close();
- } else {
- // Shut down in the non-HA case.
- try {
- listener.becomeStandby();
- } catch (Throwable t) {
- LOG.error("becomeStandby: " + incarnationId +
- " threw an unexpected exception", t);
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/LeaderStatusAdaptor.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/LeaderStatusAdaptor.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/LeaderStatusAdaptor.java
deleted file mode 100644
index e87b3d1..0000000
--- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/LeaderStatusAdaptor.java
+++ /dev/null
@@ -1,207 +0,0 @@
-/*
- * 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.
- */
-package org.apache.sentry.service.thrift;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.curator.framework.CuratorFramework;
-import org.apache.curator.framework.CuratorFrameworkFactory;
-import org.apache.curator.framework.recipes.leader.LeaderSelector;
-import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
-import org.apache.curator.retry.ExponentialBackoffRetry;
-import org.apache.hadoop.conf.Configuration;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.ReentrantLock;
-
-import static org.apache.sentry.service.thrift.ServiceConstants.ServerConfig.SENTRY_HA_ZOOKEEPER_QUORUM;
-import static org.apache.sentry.service.thrift.ServiceConstants.ServerConfig.SENTRY_HA_ZOOKEEPER_QUORUM_DEFAULT;
-import static org.apache.sentry.service.thrift.ServiceConstants.ServerConfig.SENTRY_HA_ZOOKEEPER_NAMESPACE;
-import static org.apache.sentry.service.thrift.ServiceConstants.ServerConfig.SENTRY_HA_ZOOKEEPER_NAMESPACE_DEFAULT;
-
-/**
- * Determines the leadership status of the Sentry daemon.
- */
-final class LeaderStatusAdaptor
- extends LeaderSelectorListenerAdapter implements Closeable {
- private static final Log LOG =
- LogFactory.getLog(LeaderStatusAdaptor.class);
-
- private final String LEADER_SELECTOR_SUFFIX = "leader";
-
- /**
- * The ZooKeeper path prefix to use.
- */
- private final String zkNamespace;
-
- /**
- * The Curator framework object.
- */
- private final CuratorFramework framework;
-
- /**
- * The listener which we should notify about HA state changes.
- */
- private final LeaderStatus.Listener listener;
-
- /**
- * The Curator LeaderSelector object.
- */
- private final LeaderSelector leaderSelector;
-
- /**
- * The lock which protects isActive.
- */
- private final ReentrantLock lock = new ReentrantLock();
-
- /**
- * A condition variable which the takeLeadership function will wait on.
- */
- private final Condition cond = lock.newCondition();
-
- /**
- * The number of times this incarnation has become the leader.
- */
- private long becomeLeaderCount = 0;
-
- /**
- * True only if this LeaderStatusAdaptor is closed.
- */
- private boolean isClosed = false;
-
- /**
- * True only if this incarnation is currently active.
- */
- private boolean isActive = false;
-
- LeaderStatusAdaptor(String incarnationId, Configuration conf,
- LeaderStatus.Listener listener) {
- this.zkNamespace = conf.get(SENTRY_HA_ZOOKEEPER_NAMESPACE,
- SENTRY_HA_ZOOKEEPER_NAMESPACE_DEFAULT);
- String zkServers = conf.get(SENTRY_HA_ZOOKEEPER_QUORUM,
- SENTRY_HA_ZOOKEEPER_QUORUM_DEFAULT);
- if ((zkServers == null) || (zkServers.trim().isEmpty())) {
- throw new RuntimeException("You must configure some ZooKeeper " +
- "servers via " + SENTRY_HA_ZOOKEEPER_QUORUM + " when enabling HA");
- }
- this.framework = CuratorFrameworkFactory.newClient(zkServers,
- new ExponentialBackoffRetry(1000, 3));
- this.framework.start();
- this.listener = listener;
- this.leaderSelector = new LeaderSelector(this.framework,
- this.zkNamespace + "/" + LEADER_SELECTOR_SUFFIX, this);
- this.leaderSelector.setId(incarnationId);
- this.leaderSelector.autoRequeue();
- LOG.info("Created LeaderStatusAdaptor(zkNamespace=" + zkNamespace +
- ", incarnationId=" + incarnationId +
- ", zkServers='" + zkServers + "')");
- }
-
- public void start() {
- this.leaderSelector.start();
- }
-
- /**
- * Shut down the LeaderStatusAdaptor and wait for it to transition to
- * standby.
- */
- @Override
- public void close() throws IOException {
- // If the adaptor is already closed, calling close again is a no-op.
- // Setting isClosed also prevents activation after this point.
- lock.lock();
- try {
- if (isClosed) {
- return;
- }
- isClosed = true;
- } finally {
- lock.unlock();
- }
-
- // Shut down our Curator hooks.
- leaderSelector.close();
-
- // Wait for the adaptor to transition to standby state.
- lock.lock();
- try {
- while (isActive) {
- cond.awaitUninterruptibly();
- }
- } finally {
- lock.unlock();
- }
- }
-
- /**
- * @return true if this client is the current leader.
- */
- public boolean isActive() {
- lock.lock();
- try {
- return isActive;
- } finally {
- lock.unlock();
- }
- }
-
- /**
- * Deactivate the current client, if it is active.
- */
- public void deactivate() {
- lock.lock();
- try {
- if (isActive) {
- isActive = false;
- cond.signal();
- }
- } finally {
- lock.unlock();
- }
- }
-
- @Override
- public void takeLeadership(CuratorFramework client) throws Exception {
- lock.lock();
- try {
- if (isClosed) {
- LOG.info("LeaderStatusAdaptor: can't become active because the " +
- "adaptor is closed.");
- return;
- }
- isActive = true;
- becomeLeaderCount++;
- LOG.info("LeaderStatusAdaptor: becoming active. " +
- "becomeLeaderCount=" + becomeLeaderCount);
- listener.becomeActive();
- while (isActive) {
- cond.await();
- }
- } finally {
- deactivate();
- LOG.info("LeaderStatusAdaptor: becoming standby");
- try {
- listener.becomeStandby();
- } catch (Throwable t) {
- LOG.error("becomeStandby threw unexpected exception", t);
- }
- lock.unlock();
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/LeaderStatusMonitor.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/LeaderStatusMonitor.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/LeaderStatusMonitor.java
new file mode 100644
index 0000000..d1999b2
--- /dev/null
+++ b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/LeaderStatusMonitor.java
@@ -0,0 +1,270 @@
+/*
+ * 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.
+ */
+package org.apache.sentry.service.thrift;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.recipes.leader.LeaderSelector;
+import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
+import org.apache.curator.framework.recipes.leader.LeaderSelectorListenerAdapter;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.sentry.provider.db.service.persistent.HAContext;
+
+import javax.annotation.concurrent.GuardedBy;
+import javax.annotation.concurrent.ThreadSafe;
+import java.lang.management.ManagementFactory;
+import java.lang.management.RuntimeMXBean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import static org.apache.sentry.service.thrift.ServiceConstants.ServerConfig.*;
+
+/**
+ * LeaderStatusMonitor participates in the distributed leader election protocol
+ * and allows clients to access the global leaadership status.
+ * <p>
+ * LeaderStatusMonitor is a singleton that uses Curator framework via
+ * {@link HAContext}.The leadership status can be accessed via the
+ * {@link #isLeader()} method.<p>
+ *
+ * Usually leadership re-election is initiated by the Curator framework when one
+ * of the nodes disconnects from ZooKeeper, but LeaderStatusMonitor also supports
+ * voluntary release of the leadership via the {@link #deactivate()} method. This is
+ * intended to be used for debugging purposes.
+ * <p>
+ * The class also simulates leader election in non-HA environments. In such cases its
+ * {@link #isLeader()} method always returns True. The non-HA environment is determined
+ * by the absence of the SENTRY_HA_ZOOKEEPER_QUORUM in the configuration.
+ *
+ * <h2>Implementation notes</h2>
+ *
+ * <h3>Initialization</h3>
+ *
+ * Class initialization is split between the constructor and the {@link #init()} method.
+ * There are two reasons for it:
+ * <ul>
+ * <li>We do not want to pass <strong>this</strong> reference to
+ * {@link HAContext#newLeaderSelector(String, LeaderSelectorListener)}
+ * until it is fully initialized</li>
+ * <li>We do not want to call {@link LeaderSelector#start()} method in constructor</li>
+ * </ul>
+ *
+ * Since LeaderStatusMonitor is a singleton and an instance can only be obtained via the
+ * {@link #getLeaderStatusMonitor(Configuration)} method, we hide this construction split
+ * from the callers.
+ *
+ * <h3>Synchronization</h3>
+ * Singleton synchronization is achieved using the synchronized class builder
+ * {@link #getLeaderStatusMonitor(Configuration)}
+ * <p>
+ * Upon becoming a leader, the code loops in {@link #takeLeadership(CuratorFramework)}
+ * until it receives a deactivation signal from {@link #deactivate()}. This is synchronized
+ * using a {@link #lock} and condition variable {@link #cond}.
+ * <p>
+ * Access to the leadership status {@link #isLeader} is also protected by the {@link #lock}.
+ * This isn't strictly necessary and a volatile field would be sufficient, but since we
+ * already use the {@link #lock} this is more straightforward.
+ */
+@ThreadSafe
+final class LeaderStatusMonitor
+ extends LeaderSelectorListenerAdapter implements AutoCloseable {
+
+ private static final Log LOG =
+ LogFactory.getLog(LeaderStatusMonitor.class);
+
+ private static final String LEADER_SELECTOR_SUFFIX = "leader";
+
+ /** Unique instance of the singleton object */
+ private static LeaderStatusMonitor leaderStatusMonitor = null;
+
+ private final String zkNamespace;
+ private final HAContext haContext;
+
+ /** Unique string describing this instance */
+ private final String incarnationId = generateIncarnationId();
+
+ /** True when not using ZooKeeeper */
+ private final boolean isSingleNodeMode;
+
+ /** Lock and condition used to signal the leader to voluntary release leadership */
+ private final Lock lock = new ReentrantLock();
+ /** Condition variable used to synchronize voluntary leadership release */
+ private final Condition cond = lock.newCondition();
+ /** Leadership status - true if leader. */
+ @GuardedBy("lock")
+ private boolean isLeader = false;
+
+ /** Curator framework leader monitor */
+ private LeaderSelector leaderSelector = null;
+
+ /** The number of times this incarnation has become the leader. */
+ private final AtomicLong leaderCount = new AtomicLong(0);
+
+ /**
+ * Constructor. Initialize state and create HA context if configuration
+ * specifies ZooKeeper servers.
+ * @param conf Configuration. The fields we are interested in are:
+ * <ul>
+ * <li>SENTRY_HA_ZOOKEEPER_NAMESPACE</li>
+ * <li>SENTRY_HA_ZOOKEEPER_QUORUM</li>
+ * </ul>
+ * Configuration is also passed to the
+ * {@link HAContext#newLeaderSelector(String, LeaderSelectorListener)}
+ * which uses more properties.
+ * @throws Exception
+ */
+ private LeaderStatusMonitor(Configuration conf) throws Exception {
+ // Only enable HA configuration if zookeeper is configured
+ zkNamespace = conf.get(SENTRY_HA_ZOOKEEPER_NAMESPACE, "");
+ String zkServers = conf.get(SENTRY_HA_ZOOKEEPER_QUORUM, "");
+ if (zkServers.isEmpty()) {
+ isSingleNodeMode = true;
+ haContext = null;
+ isLeader = true;
+ LOG.info("Leader election protocol disabled, assuming single active server");
+ return;
+ }
+ isSingleNodeMode = false;
+ haContext = HAContext.getHAServerContext(conf);
+
+ LOG.info("Created LeaderStatusMonitor(zkNamespace=" + zkNamespace +
+ ", incarnationId=" + incarnationId +
+ ", zkServers='" + zkServers + "')");
+ }
+
+ /**
+ * Second half of the constructor links this object with {@link HAContext} and
+ * starts leader election protocol.
+ */
+ private void init() {
+ if (isSingleNodeMode) {
+ return;
+ }
+
+ leaderSelector = haContext.newLeaderSelector(zkNamespace + "/" +
+ LEADER_SELECTOR_SUFFIX, this);
+ leaderSelector.setId(incarnationId);
+ leaderSelector.autoRequeue();
+ leaderSelector.start();
+ }
+
+ /**
+ *
+ * @param conf Configuration. See {@link #LeaderStatusMonitor(Configuration)} for details.
+ * @return A global LeaderStatusMonitor instance.
+ * @throws Exception
+ */
+ @SuppressWarnings("LawOfDemeter")
+ static synchronized LeaderStatusMonitor getLeaderStatusMonitor(Configuration conf)
+ throws Exception {
+ if (leaderStatusMonitor == null) {
+ leaderStatusMonitor = new LeaderStatusMonitor(conf);
+ leaderStatusMonitor.init();
+ }
+ return leaderStatusMonitor;
+ }
+
+ /**
+ * @return number of times this leader was elected. Used for metrics.
+ */
+ long getLeaderCount() {
+ return leaderCount.get();
+ }
+
+ /**
+ * Shut down the LeaderStatusMonitor and wait for it to transition to
+ * standby.
+ */
+ @Override
+ public void close() {
+ if (leaderSelector != null) {
+ // Shut down our Curator hooks.
+ leaderSelector.close();
+ }
+ }
+
+ /**
+ * Deactivate the current client, if it is active.
+ * In non-HA case this is a no-op.
+ */
+ void deactivate() {
+ if (isSingleNodeMode) {
+ return;
+ }
+ lock.lock();
+ try {
+ cond.signal();
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * @return true iff we are the leader.
+ * In non-HA case always returns true
+ */
+ boolean isLeader() {
+ if (isSingleNodeMode) {
+ return true;
+ }
+ lock.lock();
+ @SuppressWarnings("FieldAccessNotGuarded")
+ boolean leader = isLeader;
+ lock.unlock();
+ return leader;
+ }
+
+ /**
+ * Curator framework callback which is called when we become a leader.
+ * Should return only when we decide to resign.
+ */
+ @Override
+ public void takeLeadership(CuratorFramework client) throws Exception {
+ leaderCount.incrementAndGet();
+ LOG.info("LeaderStatusMonitor: becoming active. " +
+ "leaderCount=" + leaderCount);
+ lock.lock();
+ try {
+ isLeader = true;
+ // Wait until we are interrupted or receive a signal
+ cond.await();
+ } catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
+ LOG.info("LeaderStatusMonitor: interrupted");
+ } finally {
+ isLeader = false;
+ lock.unlock();
+ LOG.info("LeaderStatusMonitor: becoming standby");
+ }
+ }
+
+ /**
+ * Generate ID for the activator. <p>
+ *
+ * Ideally we would like something like host@pid, but Java doesn't provide a good
+ * way to determine pid value, so we use
+ * {@link RuntimeMXBean#getName()} which usually contains host
+ * name and pid.
+ */
+ private static String generateIncarnationId() {
+ return ManagementFactory.getRuntimeMXBean().getName();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/SentryService.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/SentryService.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/SentryService.java
index a9d49f8..bd4ada4 100644
--- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/SentryService.java
+++ b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/SentryService.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -45,7 +45,7 @@ import org.apache.hadoop.security.SaslRpcServer;
import org.apache.hadoop.security.SaslRpcServer.AuthMethod;
import org.apache.hadoop.security.SecurityUtil;
import org.apache.sentry.Command;
-import org.apache.sentry.core.common.utils.SentryConstants;
+import org.apache.sentry.core.common.utils.SigUtils;
import org.apache.sentry.provider.db.service.thrift.SentryHealthCheckServletContextListener;
import org.apache.sentry.provider.db.service.thrift.SentryMetrics;
import org.apache.sentry.provider.db.service.thrift.SentryMetricsServletContextListener;
@@ -67,20 +67,22 @@ import org.slf4j.LoggerFactory;
import com.google.common.base.Preconditions;
-public class SentryService implements Callable {
+import static org.apache.sentry.core.common.utils.SigUtils.registerSigListener;
- private static final Logger LOGGER = LoggerFactory
- .getLogger(SentryService.class);
+public class SentryService implements Callable, SigUtils.SigListener {
- private static enum Status {
- NOT_STARTED(), STARTED();
+ private static final Logger LOGGER = LoggerFactory.getLogger(SentryService.class);
+
+ private enum Status {
+ NOT_STARTED,
+ STARTED,
}
private final Configuration conf;
private final InetSocketAddress address;
private final int maxThreads;
private final int minThreads;
- private boolean kerberos;
+ private final boolean kerberos;
private final String principal;
private final String[] principalParts;
private final String keytab;
@@ -89,12 +91,10 @@ public class SentryService implements Callable {
private Future serviceStatus;
private TServer thriftServer;
private Status status;
- private int webServerPort;
+ private final int webServerPort;
private SentryWebServer sentryWebServer;
- private long maxMessageSize;
- private final boolean isHA;
- private final Activator act;
- SentryMetrics sentryMetrics;
+ private final long maxMessageSize;
+ private final LeaderStatusMonitor leaderMonitor;
public SentryService(Configuration conf) throws Exception {
this.conf = conf;
@@ -139,8 +139,6 @@ public class SentryService implements Callable {
principalParts = null;
keytab = null;
}
- isHA = conf.getBoolean(ServerConfig.SENTRY_HA_ENABLED,
- ServerConfig.SENTRY_HA_ENABLED_DEFAULT);
serviceExecutor = Executors.newSingleThreadExecutor(new ThreadFactory() {
private int count = 0;
@@ -150,19 +148,28 @@ public class SentryService implements Callable {
+ (count++));
}
});
- this.act = Activators.INSTANCE.create(conf);
- conf.set(SentryConstants.CURRENT_INCARNATION_ID_KEY,
- this.act.getIncarnationId());
+ this.leaderMonitor = LeaderStatusMonitor.getLeaderStatusMonitor(conf);
webServerPort = conf.getInt(ServerConfig.SENTRY_WEB_PORT, ServerConfig.SENTRY_WEB_PORT_DEFAULT);
- //TODO: Enable only if Hive is using Sentry?
try {
hmsFollowerExecutor = Executors.newScheduledThreadPool(1);
- hmsFollowerExecutor.scheduleAtFixedRate(new HMSFollower(conf), 60000, 500, TimeUnit.MILLISECONDS);
- }catch(Exception e) {
+ hmsFollowerExecutor.scheduleAtFixedRate(new HMSFollower(conf),
+ 60000, 500, TimeUnit.MILLISECONDS);
+ } catch(Exception e) {
//TODO: Handle
LOGGER.error("Could not start HMSFollower");
}
status = Status.NOT_STARTED;
+
+ // Enable signal handler for HA leader/follower status if configured
+ String sigName = conf.get(ServerConfig.SERVER_HA_STANDBY_SIG);
+ if ((sigName != null) && !sigName.isEmpty()) {
+ LOGGER.info("Registering signal handler " + sigName + " for HA");
+ try {
+ registerSigListener(sigName, this);
+ } catch (Exception e) {
+ LOGGER.error("Failed to register signal", e);
+ }
+ }
}
@Override
@@ -251,8 +258,7 @@ public class SentryService implements Callable {
}
private void addSentryServiceGauge() {
- sentryMetrics = SentryMetrics.getInstance();
- sentryMetrics.addSentryServiceGauges(this);
+ SentryMetrics.getInstance().addSentryServiceGauges(this);
}
private void startSentryWebServer() throws Exception{
@@ -295,7 +301,7 @@ public class SentryService implements Callable {
public synchronized void stop() throws Exception{
MultiException exception = null;
LOGGER.info("Attempting to stop...");
- act.close();
+ leaderMonitor.close();
if (isRunning()) {
LOGGER.info("Attempting to stop sentry thrift service...");
try {
@@ -337,16 +343,8 @@ public class SentryService implements Callable {
* to the backend DB.
*/
@VisibleForTesting
- public synchronized void becomeStandby() throws Exception{
- try {
- if(act.isActive()) {
- LOGGER.info("Server with incarnation id: " + act.getIncarnationId() +
- " becoming standby");
- act.deactivate();
- }
- } catch (Exception e) {
- LOGGER.error("Error while deactivating the active sentry daemon", e);
- }
+ public synchronized void becomeStandby() {
+ leaderMonitor.deactivate();
}
private MultiException addMultiException(MultiException exception, Exception e) {
@@ -463,24 +461,27 @@ public class SentryService implements Callable {
return thriftServer.getEventHandler();
}
- @VisibleForTesting
- public Activator getActivator() {return act;}
-
public Gauge<Boolean> getIsActiveGauge() {
return new Gauge<Boolean>() {
@Override
public Boolean getValue() {
- return act.isActive();
+ return leaderMonitor.isLeader();
}
};
}
- public Gauge<Boolean> getIsHAGauge() {
- return new Gauge<Boolean>() {
+ public Gauge<Long> getBecomeActiveCount() {
+ return new Gauge<Long>() {
@Override
- public Boolean getValue() {
- return isHA;
+ public Long getValue() {
+ return leaderMonitor.getLeaderCount();
}
};
}
+
+ @Override
+ public void onSignal(String signalName) {
+ // Become follower
+ leaderMonitor.deactivate();
+ }
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/ServiceConstants.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/ServiceConstants.java b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/ServiceConstants.java
index d80fa1e..c4fdf1d 100644
--- a/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/ServiceConstants.java
+++ b/sentry-provider/sentry-provider-db/src/main/java/org/apache/sentry/service/thrift/ServiceConstants.java
@@ -124,18 +124,18 @@ public class ServiceConstants {
public static final String SENTRY_HA_ZOOKEEPER_SECURITY = SENTRY_HA_ZK_PROPERTY_PREFIX + "security";
public static final boolean SENTRY_HA_ZOOKEEPER_SECURITY_DEFAULT = false;
public static final String SENTRY_HA_ZOOKEEPER_QUORUM = SENTRY_HA_ZK_PROPERTY_PREFIX + "quorum";
- public static final String SENTRY_HA_ZOOKEEPER_QUORUM_DEFAULT = "";
public static final String SENTRY_HA_ZOOKEEPER_RETRIES_MAX_COUNT = SENTRY_HA_ZK_PROPERTY_PREFIX + "session.retries.max.count";
public static final int SENTRY_HA_ZOOKEEPER_RETRIES_MAX_COUNT_DEFAULT = 3;
public static final String SENTRY_HA_ZOOKEEPER_SLEEP_BETWEEN_RETRIES_MS = SENTRY_HA_ZK_PROPERTY_PREFIX + "session.sleep.between.retries.ms";
public static final int SENTRY_HA_ZOOKEEPER_SLEEP_BETWEEN_RETRIES_MS_DEFAULT = 100;
public static final String SENTRY_HA_ZOOKEEPER_NAMESPACE = SENTRY_HA_ZK_PROPERTY_PREFIX + "namespace";
- public static final String SENTRY_HA_ZOOKEEPER_NAMESPACE_DEFAULT = "/sentry";
+ public static final String SENTRY_HA_ZOOKEEPER_NAMESPACE_DEFAULT = "sentry";
// principal and keytab for client to be able to connect to secure ZK. Needed for Sentry HA with secure ZK
public static final String SERVER_HA_ZOOKEEPER_CLIENT_PRINCIPAL = "sentry.zookeeper.client.principal";
public static final String SERVER_HA_ZOOKEEPER_CLIENT_KEYTAB = "sentry.zookeeper.client.keytab";
public static final String SERVER_HA_ZOOKEEPER_CLIENT_TICKET_CACHE = "sentry.zookeeper.client.ticketcache";
public static final String SERVER_HA_ZOOKEEPER_CLIENT_TICKET_CACHE_DEFAULT = "false";
+ public static final String SERVER_HA_STANDBY_SIG = "sentry.ha.standby.signal";
public static final ImmutableMap<String, String> SENTRY_STORE_DEFAULTS =
ImmutableMap.<String, String>builder()
.put("datanucleus.connectionPoolingType", "BoneCP")
@@ -222,9 +222,7 @@ public class ServiceConstants {
// HA configuration
public static final String SENTRY_HA_ENABLED = "sentry.ha.enabled";
- public static final boolean SENTRY_HA_ENABLED_DEFAULT = ServerConfig.SENTRY_HA_ENABLED_DEFAULT;
public static final String SENTRY_HA_ZOOKEEPER_QUORUM = ServerConfig.SENTRY_HA_ZOOKEEPER_QUORUM;
- public static final String SERVER_HA_ZOOKEEPER_QUORUM_DEFAULT = ServerConfig.SENTRY_HA_ZOOKEEPER_QUORUM_DEFAULT;
public static final String SENTRY_HA_ZOOKEEPER_NAMESPACE = ServerConfig.SENTRY_HA_ZOOKEEPER_NAMESPACE;
public static final String SERVER_HA_ZOOKEEPER_NAMESPACE_DEFAULT = ServerConfig.SENTRY_HA_ZOOKEEPER_NAMESPACE_DEFAULT;
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/thrift/TestActivator.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/thrift/TestActivator.java b/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/thrift/TestActivator.java
index 60cfc73..5227c45 100644
--- a/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/thrift/TestActivator.java
+++ b/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/thrift/TestActivator.java
@@ -37,14 +37,6 @@ public class TestActivator extends SentryServiceIntegrationBase {
}
- @Test
- public void testStopActive() throws Exception {
- Assert.assertEquals(Boolean.TRUE,server.getActivator().isActive());
- //stop the current service.
- server.becomeStandby();
- Assert.assertEquals(Boolean.FALSE,server.getActivator().isActive());
- }
-
@Override
@After
public void after() {
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/thrift/TestSentryServiceMetrics.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/thrift/TestSentryServiceMetrics.java b/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/thrift/TestSentryServiceMetrics.java
index bc375e3..a33b03a 100644
--- a/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/thrift/TestSentryServiceMetrics.java
+++ b/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/provider/db/service/thrift/TestSentryServiceMetrics.java
@@ -56,7 +56,6 @@ public class TestSentryServiceMetrics extends SentryServiceIntegrationBase {
//More Cases to be added once Sentry HA is implemented
//Check for gauges with the server handle.
- Assert.assertEquals(Boolean.FALSE,server.getIsHAGauge().getValue());
Assert.assertEquals(Boolean.TRUE,server.getIsActiveGauge().getValue());
}
http://git-wip-us.apache.org/repos/asf/sentry/blob/6125ac9b/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/service/thrift/TestLeaderStatus.java
----------------------------------------------------------------------
diff --git a/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/service/thrift/TestLeaderStatus.java b/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/service/thrift/TestLeaderStatus.java
deleted file mode 100644
index c796fab..0000000
--- a/sentry-provider/sentry-provider-db/src/test/java/org/apache/sentry/service/thrift/TestLeaderStatus.java
+++ /dev/null
@@ -1,245 +0,0 @@
-/*
- * 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.
- */
-package org.apache.sentry.service.thrift;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.curator.test.TestingServer;
-import org.apache.curator.utils.CloseableUtils;
-import org.apache.hadoop.conf.Configuration;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.Semaphore;
-
-import static org.apache.sentry.service.thrift.ServiceConstants.ClientConfig.SENTRY_HA_ENABLED;
-import static org.apache.sentry.service.thrift.ServiceConstants.ClientConfig.SENTRY_HA_ZOOKEEPER_QUORUM;
-
-final public class TestLeaderStatus {
- private static final Log LOG =
- LogFactory.getLog(TestLeaderStatus.class);
-
- /**
- * Test that when the configuration is non-HA, we always become active.
- */
- @Test(timeout = 60000)
- public void testNonHaLeaderStatus() throws Exception {
- Configuration conf = new Configuration();
- conf.set(SENTRY_HA_ZOOKEEPER_QUORUM, "");
- final Semaphore activeSem = new Semaphore(0);
- final Semaphore standbySem = new Semaphore(0);
- LeaderStatus status = new LeaderStatus(new LeaderStatus.Listener() {
- @Override
- public void becomeActive() throws Exception {
- LOG.info("testNonHaLeaderStatus: becoming active");
- activeSem.release(2);
- }
-
- @Override
- public void becomeStandby() {
- activeSem.acquireUninterruptibly();
- LOG.info("testNonHaLeaderStatus: becoming standby");
- standbySem.release();
- }
- }, conf);
- status.start();
- activeSem.acquire();
- status.close();
- standbySem.acquire();
- }
-
- private static class CurrentTestActive {
- private String incarnationId;
- private String error = null;
-
- CurrentTestActive() {
- this.incarnationId = null;
- this.error = null;
- }
-
- synchronized void set(String incarnationId) {
- if (this.incarnationId != null) {
- error("set: there is already an " +
- "active incarnation " + this.incarnationId);
- return;
- }
- this.incarnationId = incarnationId;
- }
-
- synchronized void unset(String incarnationId) {
- if (this.incarnationId == null) {
- error("unset: there is no active incarnation.");
- return;
- }
- if (!this.incarnationId.equals(incarnationId)) {
- error("unset: can't deactivate " +
- incarnationId + " because " + this.incarnationId +
- " is the current active incarnation.");
- return;
- }
- this.incarnationId = null;
- }
-
- synchronized String get() {
- return this.incarnationId;
- }
-
- synchronized String getError() {
- return error;
- }
-
- synchronized void error(String error) {
- if (this.error == null) {
- this.error = error;
- }
- LOG.error(error);
- }
-
- String busyWaitForActive() throws InterruptedException {
- for (; ; ) {
- String cur = get();
- if (cur != null) {
- return cur;
- }
- Thread.sleep(2);
- }
- }
-
- String busyWaitForNextActive(String prevIncarnation)
- throws InterruptedException {
- for (; ; ) {
- String cur = get();
- if ((cur != null) && (!cur.equals(prevIncarnation))) {
- return cur;
- }
- Thread.sleep(2);
- }
- }
- }
-
- static class LeaderStatusContext implements Closeable {
- final LeaderStatus status;
-
- LeaderStatusContext(final CurrentTestActive active,
- Configuration conf) throws Exception {
- this.status = new LeaderStatus(new LeaderStatus.Listener() {
- @Override
- public void becomeActive() throws Exception {
- LOG.info("LeaderStatusContext " + status.getIncarnationId() +
- " becoming active");
- active.set(status.getIncarnationId());
- }
-
- @Override
- public void becomeStandby() {
- LOG.info("LeaderStatusContext " + status.getIncarnationId() +
- " becoming standby");
- active.unset(status.getIncarnationId());
- }
- }, conf);
- this.status.start();
- }
-
- @Override
- public void close() throws IOException {
- this.status.close();
- }
-
- @Override
- public String toString() {
- return "LeaderStatusContext(" + status.getIncarnationId() + ")";
- }
-
- String getIncarnationId() {
- return status.getIncarnationId();
- }
- }
-
- @Test(timeout = 120000)
- public void testRacingClients() throws Exception {
- final int NUM_CLIENTS = 3;
- final Configuration conf = new Configuration();
- TestingServer server = new TestingServer();
- server.start();
- conf.setBoolean(SENTRY_HA_ENABLED, true);
- conf.set(SENTRY_HA_ZOOKEEPER_QUORUM, server.getConnectString());
- final CurrentTestActive active = new CurrentTestActive();
- List<LeaderStatusContext> contexts = new LinkedList<>();
- for (int i = 0; i < NUM_CLIENTS; i++) {
- try {
- contexts.add(new LeaderStatusContext(active, conf));
- } catch (Throwable t) {
- LOG.error("error creating LeaderStatusContext", t);
- throw new RuntimeException(t);
- }
- }
- LOG.info("Created " + NUM_CLIENTS + " SentryLeaderSelectorClient " +
- "objects.");
- String curIncarnation = active.busyWaitForActive();
- LOG.info("Closing LeaderStatus(" + curIncarnation + ").");
- for (Iterator<LeaderStatusContext> iter = contexts.iterator();
- iter.hasNext(); ) {
- LeaderStatusContext context = iter.next();
- if (context.getIncarnationId().equals(curIncarnation)) {
- CloseableUtils.closeQuietly(context);
- iter.remove();
- }
- }
- active.busyWaitForNextActive(curIncarnation);
- for (Iterator<LeaderStatusContext> iter = contexts.iterator();
- iter.hasNext(); ) {
- LeaderStatusContext context = iter.next();
- CloseableUtils.closeQuietly(context);
- iter.remove();
- }
- LOG.info("Closed all " + NUM_CLIENTS + " SentryLeaderSelectorClient " +
- "objects.");
- Assert.assertTrue(null == active.getError());
- server.close();
- }
-
- @Test(timeout = 60000)
- public void testGenerateIncarnationIDs() throws Exception {
- final int NUM_UNIQUE_IDS = 10000;
- HashSet<String> ids = new HashSet<String>();
- for (int i = 0; i < NUM_UNIQUE_IDS; i++) {
- ids.add(LeaderStatus.generateIncarnationId());
- }
-
- // Assert that there were no ID collisions
- Assert.assertEquals(NUM_UNIQUE_IDS, ids.size());
-
- // Assert that all IDs are 16 characters long and begin with a letter.
- for (String id : ids) {
- Assert.assertEquals(16, id.length());
- Assert.assertTrue(Character.isAlphabetic(id.charAt(0)));
- }
-
- // Assert that IDs contain only alphanumeric characters
- for (String id : ids) {
- for (int i = 0; i < id.length(); i++) {
- Assert.assertTrue(Character.isLetterOrDigit(id.charAt(i)));
- }
- }
- }
-}