You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sentry.apache.org by sp...@apache.org on 2018/05/31 03:32:38 UTC
[60/86] sentry git commit: SENTRY-2208: Refactor out Sentry service
into own module from sentry-provider-db (Anthony Young-Garner,
reviewed by Sergio Pena, Steve Moist, Na Li)
http://git-wip-us.apache.org/repos/asf/sentry/blob/7db84b2f/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryPrivilege.java
----------------------------------------------------------------------
diff --git a/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryPrivilege.java b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryPrivilege.java
new file mode 100644
index 0000000..85477b6
--- /dev/null
+++ b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryPrivilege.java
@@ -0,0 +1,347 @@
+/**
+ * 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.model;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import javax.jdo.annotations.PersistenceCapable;
+
+import org.apache.sentry.core.common.utils.PathUtils;
+import org.apache.sentry.core.model.db.AccessConstants;
+import org.apache.sentry.provider.db.service.persistent.SentryStore;
+
+/**
+ * Database backed Sentry Privilege. Any changes to this object
+ * require re-running the maven build so DN an re-enhance.
+ */
+@PersistenceCapable
+public class MSentryPrivilege {
+
+ private String privilegeScope;
+ /**
+ * Privilege name is unique
+ */
+ private String serverName = "";
+ private String dbName = "";
+ private String tableName = "";
+ private String columnName = "";
+ private String URI = "";
+ private String action = "";
+ private Boolean grantOption = false;
+ // roles this privilege is a part of
+ private Set<MSentryRole> roles;
+ // users this privilege is a part of
+ private Set<MSentryUser> users;
+ private long createTime;
+
+ public MSentryPrivilege() {
+ this.roles = new HashSet<>();
+ this.users = new HashSet<>();
+ }
+
+ public MSentryPrivilege(String privilegeScope,
+ String serverName, String dbName, String tableName, String columnName,
+ String URI, String action, Boolean grantOption) {
+ this.privilegeScope = MSentryUtil.safeIntern(privilegeScope);
+ this.serverName = MSentryUtil.safeIntern(serverName);
+ this.dbName = SentryStore.toNULLCol(dbName).intern();
+ this.tableName = SentryStore.toNULLCol(tableName).intern();
+ this.columnName = SentryStore.toNULLCol(columnName).intern();
+ this.URI = SentryStore.toNULLCol(URI).intern();
+ this.action = SentryStore.toNULLCol(action).intern();
+ this.grantOption = grantOption;
+ this.roles = new HashSet<>();
+ this.users = new HashSet<>();
+ }
+
+ public MSentryPrivilege(String privilegeScope,
+ String serverName, String dbName, String tableName, String columnName,
+ String URI, String action) {
+ this(privilegeScope, serverName, dbName, tableName,
+ columnName, URI, action, false);
+ }
+
+ public MSentryPrivilege(MSentryPrivilege other) {
+ this.privilegeScope = other.privilegeScope;
+ this.serverName = other.serverName;
+ this.dbName = SentryStore.toNULLCol(other.dbName).intern();
+ this.tableName = SentryStore.toNULLCol(other.tableName).intern();
+ this.columnName = SentryStore.toNULLCol(other.columnName).intern();
+ this.URI = SentryStore.toNULLCol(other.URI).intern();
+ this.action = SentryStore.toNULLCol(other.action).intern();
+ this.grantOption = other.grantOption;
+ this.roles = new HashSet<>();
+ roles.addAll(other.roles);
+ this.users = new HashSet<>();
+ users.addAll(other.users);
+ }
+
+ public String getServerName() {
+ return serverName;
+ }
+
+ public void setServerName(String serverName) {
+ this.serverName = (serverName == null) ? "" : serverName;
+ }
+
+ public String getDbName() {
+ return dbName;
+ }
+
+ public void setDbName(String dbName) {
+ this.dbName = (dbName == null) ? "" : dbName;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ public void setTableName(String tableName) {
+ this.tableName = (tableName == null) ? "" : tableName;
+ }
+
+ public String getColumnName() {
+ return columnName;
+ }
+
+ public void setColumnName(String columnName) {
+ this.columnName = (columnName == null) ? "" : columnName;
+ }
+
+ public String getURI() {
+ return URI;
+ }
+
+ public void setURI(String uRI) {
+ URI = (uRI == null) ? "" : uRI;
+ }
+
+ public String getAction() {
+ return action;
+ }
+
+ public void setAction(String action) {
+ this.action = (action == null) ? "" : action;
+ }
+
+ public long getCreateTime() {
+ return createTime;
+ }
+
+ public void setCreateTime(long createTime) {
+ this.createTime = createTime;
+ }
+
+ public String getPrivilegeScope() {
+ return privilegeScope;
+ }
+
+ public void setPrivilegeScope(String privilegeScope) {
+ this.privilegeScope = privilegeScope;
+ }
+
+ public Boolean getGrantOption() {
+ return grantOption;
+ }
+
+ public void setGrantOption(Boolean grantOption) {
+ this.grantOption = grantOption;
+ }
+
+ public void appendRole(MSentryRole role) {
+ roles.add(role);
+ }
+
+ public void appendUser(MSentryUser user) {
+ users.add(user);
+ }
+
+ public Set<MSentryRole> getRoles() {
+ return roles;
+ }
+
+ public Set<MSentryUser> getUsers() { return users; }
+
+ public void removeRole(MSentryRole role) {
+ roles.remove(role);
+ role.removePrivilege(this);
+ }
+
+ public void removeUser(MSentryUser user) {
+ users.remove(user);
+ user.removePrivilege(this);
+ }
+
+ @Override
+ public String toString() {
+ return "MSentryPrivilege [privilegeScope=" + privilegeScope
+ + ", serverName=" + serverName + ", dbName=" + dbName
+ + ", tableName=" + tableName + ", columnName=" + columnName
+ + ", URI=" + URI + ", action=" + action + ", roles=[...]" + ", users=[...]"
+ + ", createTime=" + createTime + ", grantOption=" + grantOption +"]";
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((URI == null) ? 0 : URI.hashCode());
+ result = prime * result + ((action == null) ? 0 : action.hashCode());
+ result = prime * result + ((dbName == null) ? 0 : dbName.hashCode());
+ result = prime * result
+ + ((serverName == null) ? 0 : serverName.hashCode());
+ result = prime * result + ((tableName == null) ? 0 : tableName.hashCode());
+ result = prime * result
+ + ((columnName == null) ? 0 : columnName.hashCode());
+ result = prime * result
+ + ((grantOption == null) ? 0 : grantOption.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null) {
+ return false;
+ }
+ if (getClass() != obj.getClass()) {
+ return false;
+ }
+ MSentryPrivilege other = (MSentryPrivilege) obj;
+ if (URI == null) {
+ if (other.URI != null) {
+ return false;
+ }
+ } else if (!URI.equals(other.URI)) {
+ return false;
+ }
+ if (action == null) {
+ if (other.action != null) {
+ return false;
+ }
+ } else if (!action.equals(other.action)) {
+ return false;
+ }
+ if (dbName == null) {
+ if (other.dbName != null) {
+ return false;
+ }
+ } else if (!dbName.equals(other.dbName)) {
+ return false;
+ }
+ if (serverName == null) {
+ if (other.serverName != null) {
+ return false;
+ }
+ } else if (!serverName.equals(other.serverName)) {
+ return false;
+ }
+ if (tableName == null) {
+ if (other.tableName != null) {
+ return false;
+ }
+ } else if (!tableName.equals(other.tableName)) {
+ return false;
+ }
+ if (columnName == null) {
+ if (other.columnName != null) {
+ return false;
+ }
+ } else if (!columnName.equals(other.columnName)) {
+ return false;
+ }
+ if (grantOption == null) {
+ if (other.grantOption != null) {
+ return false;
+ }
+ } else if (!grantOption.equals(other.grantOption)) {
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Return true if this privilege implies other privilege
+ * Otherwise, return false
+ * @param other, other privilege
+ */
+ public boolean implies(MSentryPrivilege other) {
+ // serverName never be null
+ if (isNULL(serverName) || isNULL(other.serverName)) {
+ return false;
+ } else if (!serverName.equals(other.serverName)) {
+ return false;
+ }
+
+ // check URI implies
+ if (!isNULL(URI) && !isNULL(other.URI)) {
+ if (!PathUtils.impliesURI(URI, other.URI)) {
+ return false;
+ }
+ // if URI is NULL, check dbName and tableName
+ } else if (isNULL(URI) && isNULL(other.URI)) {
+ if (!isNULL(dbName)) {
+ if (isNULL(other.dbName)) {
+ return false;
+ } else if (!dbName.equals(other.dbName)) {
+ return false;
+ }
+ }
+ if (!isNULL(tableName)) {
+ if (isNULL(other.tableName)) {
+ return false;
+ } else if (!tableName.equals(other.tableName)) {
+ return false;
+ }
+ }
+ if (!isNULL(columnName)) {
+ if (isNULL(other.columnName)) {
+ return false;
+ } else if (!columnName.equals(other.columnName)) {
+ return false;
+ }
+ }
+ // if URI is not NULL, but other's URI is NULL, return false
+ } else if (!isNULL(URI) && isNULL(other.URI)){
+ return false;
+ }
+
+ // check action implies
+ if (!action.equalsIgnoreCase(AccessConstants.ALL)
+ && !action.equalsIgnoreCase(other.action)
+ && !action.equalsIgnoreCase(AccessConstants.ACTION_ALL)) {
+ return false;
+ }
+
+ return true;
+ }
+
+ private boolean isNULL(String s) {
+ return SentryStore.isNULL(s);
+ }
+
+ public boolean isActionALL() {
+ return AccessConstants.ACTION_ALL.equalsIgnoreCase(action)
+ || AccessConstants.ALL.equals(action);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/7db84b2f/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryRole.java
----------------------------------------------------------------------
diff --git a/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryRole.java b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryRole.java
new file mode 100644
index 0000000..fb8f5d2
--- /dev/null
+++ b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryRole.java
@@ -0,0 +1,224 @@
+/**
+ * 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.model;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import javax.jdo.annotations.PersistenceCapable;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableSet;
+
+/**
+ * Database backed Sentry Role. Any changes to this object
+ * require re-running the maven build so DN an re-enhance.
+ */
+@PersistenceCapable
+public class MSentryRole {
+
+ private String roleName;
+ // set of privileges granted to this role
+ private Set<MSentryPrivilege> privileges;
+ // set of generic model privileges grant ro this role
+ private Set<MSentryGMPrivilege> gmPrivileges;
+
+ // set of groups this role belongs to
+ private Set<MSentryGroup> groups;
+ // set of users this role belongs to
+ private Set<MSentryUser> users;
+ private long createTime;
+
+ public MSentryRole(String roleName, long createTime) {
+ this.roleName = MSentryUtil.safeIntern(roleName);
+ this.createTime = createTime;
+ privileges = new HashSet<>();
+ gmPrivileges = new HashSet<>();
+ groups = new HashSet<>();
+ users = new HashSet<>();
+ }
+
+ public MSentryRole(String roleName) {
+ this(roleName, System.currentTimeMillis());
+ }
+
+ public long getCreateTime() {
+ return createTime;
+ }
+
+ public void setCreateTime(long createTime) {
+ this.createTime = createTime;
+ }
+
+ public String getRoleName() {
+ return roleName;
+ }
+
+ public void setRoleName(String roleName) {
+ this.roleName = roleName;
+ }
+
+ public void setPrivileges(Set<MSentryPrivilege> privileges) {
+ this.privileges = privileges;
+ }
+
+ public Set<MSentryPrivilege> getPrivileges() {
+ return privileges;
+ }
+
+ public Set<MSentryGMPrivilege> getGmPrivileges() {
+ return gmPrivileges;
+ }
+
+ public void setGmPrivileges(Set<MSentryGMPrivilege> gmPrivileges) {
+ this.gmPrivileges = gmPrivileges;
+ }
+
+ public void setGroups(Set<MSentryGroup> groups) {
+ this.groups = groups;
+ }
+
+ public Set<MSentryGroup> getGroups() {
+ return groups;
+ }
+
+ public Set<MSentryUser> getUsers() {
+ return users;
+ }
+
+ public void setUsers(Set<MSentryUser> users) {
+ this.users = users;
+ }
+
+ public void removePrivilege(MSentryPrivilege privilege) {
+ if (privileges.remove(privilege)) {
+ privilege.removeRole(this);
+ }
+ }
+
+ public void appendPrivileges(Set<MSentryPrivilege> privileges) {
+ this.privileges.addAll(privileges);
+ }
+
+ public void appendPrivilege(MSentryPrivilege privilege) {
+ if (privileges.add(privilege)) {
+ privilege.appendRole(this);
+ }
+ }
+
+ public void removeGMPrivilege(MSentryGMPrivilege gmPrivilege) {
+ if (gmPrivileges.remove(gmPrivilege)) {
+ gmPrivilege.removeRole(this);
+ }
+ }
+
+ public void appendGMPrivilege(MSentryGMPrivilege gmPrivilege) {
+ if (gmPrivileges.add(gmPrivilege)) {
+ gmPrivilege.appendRole(this);
+ }
+ }
+
+ public void removeGMPrivileges() {
+ for (MSentryGMPrivilege privilege : ImmutableSet.copyOf(gmPrivileges)) {
+ privilege.removeRole(this);
+ }
+ Preconditions.checkState(gmPrivileges.isEmpty(), "gmPrivileges should be empty: " + gmPrivileges);
+ }
+
+ public void appendGroups(Set<MSentryGroup> groups) {
+ this.groups.addAll(groups);
+ }
+
+ public void appendGroup(MSentryGroup group) {
+ if (groups.add(group)) {
+ group.appendRole(this);
+ }
+ }
+
+ public void removeGroup(MSentryGroup group) {
+ if (groups.remove(group)) {
+ group.removeRole(this);
+ }
+ }
+
+ public void appendUsers(Set<MSentryUser> users) {
+ this.users.addAll(users);
+ }
+
+ public void appendUser(MSentryUser user) {
+ if (users.add(user)) {
+ user.appendRole(this);
+ }
+ }
+
+ public void removeUser(MSentryUser user) {
+ if (users.remove(user)) {
+ user.removeRole(this);
+ }
+ }
+
+ public void removePrivileges() {
+ // As we iterate through the loop below Method removeRole will modify the privileges set
+ // will be updated.
+ // Copy of the <code>privileges<code> is taken at the beginning of the loop to avoid using
+ // the actual privilege set in MSentryRole instance.
+
+ for (MSentryPrivilege privilege : ImmutableSet.copyOf(privileges)) {
+ privilege.removeRole(this);
+ }
+ Preconditions.checkState(privileges.isEmpty(), "Privileges should be empty: " + privileges);
+ }
+
+ @Override
+ public String toString() {
+ return "MSentryRole [roleName=" + roleName + ", privileges=[..]" + ", gmPrivileges=[..]"
+ + ", groups=[...]" + ", users=[...]" + ", createTime=" + createTime + "]";
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((roleName == null) ? 0 : roleName.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null) {
+ return false;
+ }
+ if (getClass() != obj.getClass()) {
+ return false;
+ }
+ MSentryRole other = (MSentryRole) obj;
+ if (roleName == null) {
+ if (other.roleName != null) {
+ return false;
+ }
+ } else if (!roleName.equals(other.roleName)) {
+ return false;
+ }
+ return true;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/7db84b2f/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryUser.java
----------------------------------------------------------------------
diff --git a/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryUser.java b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryUser.java
new file mode 100644
index 0000000..9188738
--- /dev/null
+++ b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryUser.java
@@ -0,0 +1,154 @@
+/**
+ * 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.model;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableSet;
+import java.util.HashSet;
+import java.util.Set;
+
+import javax.jdo.annotations.PersistenceCapable;
+
+/**
+ * Database backed Sentry User. Any changes to this object
+ * require re-running the maven build so DN an re-enhance.
+ */
+@PersistenceCapable
+public class MSentryUser {
+
+ /**
+ * User name is unique
+ */
+ private String userName;
+ // set of roles granted to this user
+ private Set<MSentryRole> roles;
+ // set of privileges granted to this user
+ private Set<MSentryPrivilege> privileges;
+ private long createTime;
+
+ public MSentryUser(String userName, long createTime, Set<MSentryRole> roles) {
+ this.userName = MSentryUtil.safeIntern(userName);
+ this.createTime = createTime;
+ this.roles = roles;
+ this.privileges = new HashSet<>();
+ }
+
+ public long getCreateTime() {
+ return createTime;
+ }
+
+ public void setCreateTime(long createTime) {
+ this.createTime = createTime;
+ }
+
+ public Set<MSentryRole> getRoles() {
+ return roles;
+ }
+
+ public String getUserName() {
+ return userName;
+ }
+
+ public void appendRole(MSentryRole role) {
+ if (roles.add(role)) {
+ role.appendUser(this);
+ }
+ }
+
+ public void removeRole(MSentryRole role) {
+ if (roles.remove(role)) {
+ role.removeUser(this);
+ }
+ }
+
+ public void setPrivileges(Set<MSentryPrivilege> privileges) {
+ this.privileges = privileges;
+ }
+
+ public Set<MSentryPrivilege> getPrivileges() {
+ return privileges;
+ }
+
+ public void removePrivilege(MSentryPrivilege privilege) {
+ if (privileges.remove(privilege)) {
+ privilege.removeUser(this);
+ }
+ }
+
+ public void appendPrivileges(Set<MSentryPrivilege> privileges) {
+ this.privileges.addAll(privileges);
+ }
+
+ public void appendPrivilege(MSentryPrivilege privilege) {
+ if (privileges.add(privilege)) {
+ privilege.appendUser(this);
+ }
+ }
+
+ public void removePrivileges() {
+ // As we iterate through the loop below Method removeRole will modify the privileges set
+ // will be updated.
+ // Copy of the <code>privileges<code> is taken at the beginning of the loop to avoid using
+ // the actual privilege set in MSentryUser instance.
+
+ for (MSentryPrivilege privilege : ImmutableSet.copyOf(privileges)) {
+ privilege.removeUser(this);
+ }
+ Preconditions.checkState(privileges.isEmpty(), "Privileges should be empty: " + privileges);
+ }
+
+ @Override
+ public String toString() {
+ return "MSentryUser [userName=" + userName + ", roles=[...]" + ", privileges=[...]" + ", createTime=" + createTime
+ + "]";
+ }
+
+ @Override
+ public int hashCode() {
+ final int prime = 31;
+ int result = 1;
+ result = prime * result + ((userName == null) ? 0 : userName.hashCode());
+ return result;
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null) {
+ return false;
+ }
+ if (getClass() != obj.getClass()) {
+ return false;
+ }
+ MSentryUser other = (MSentryUser) obj;
+ if (createTime != other.createTime) {
+ return false;
+ }
+ if (userName == null) {
+ if (other.userName != null) {
+ return false;
+ }
+ } else if (!userName.equals(other.userName)) {
+ return false;
+ }
+ return true;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/sentry/blob/7db84b2f/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryUtil.java
----------------------------------------------------------------------
diff --git a/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryUtil.java b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryUtil.java
new file mode 100644
index 0000000..939bf83
--- /dev/null
+++ b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryUtil.java
@@ -0,0 +1,89 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.model;
+
+import org.apache.sentry.core.common.utils.SentryUtils;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Common utilities for model objects
+ */
+public final class MSentryUtil {
+ /**
+ * Intern a string but only if it is not null
+ * @param arg String to be interned, may be null
+ * @return interned string or null
+ */
+ static String safeIntern(String arg) {
+ return (arg != null) ? arg.intern() : null;
+ }
+
+ /**
+ * Given a collection of MSentryChange's, retrieve the change id's associated and return as a list
+ * <p>
+ * e.g:
+ * <li> Input: [MSentryChange(1), MSentryChange(2), MSentryChange(3), MSentryChange(5), MSentryChange(7)] </li>
+ * <li> Output: [1, 2, 3, 5 ,7] </li>
+ * </p>
+ * @param changes List of {@link MSentryChange}
+ * @return List of changeID's
+ */
+ private static List<Long> getChangeIds(Collection<? extends MSentryChange> changes) {
+ List<Long> ids = changes.isEmpty() ? Collections.<Long>emptyList() : new ArrayList<Long>(changes.size());
+ for (MSentryChange change : changes) {
+ ids.add(change.getChangeID());
+ }
+ return ids;
+ }
+
+ /**
+ * Given a collection of MSentryChange instances sorted by ID return true if and only if IDs are sequential (do not contain holes)
+ * <p>
+ * e.g:
+ * <li> Input: [MSentryChange(1), MSentryChange(2), MSentryChange(3), MSentryChange(5), MSentryChange(7)] </li>
+ * <li> Output: False </li>
+ * <li> Input: [MSentryChange(1), MSentryChange(2), MSentryChange(3), MSentryChange(4), MSentryChange(5)] </li>
+ * <li> Output: True </li>
+ * </p>
+ * @param changes List of {@link MSentryChange}
+ * @return True if all the ids are sequential otherwise returns False
+ */
+ public static boolean isConsecutive(List<? extends MSentryChange> changes) {
+ int size = changes.size();
+ return (size <= 1) || (changes.get(size - 1).getChangeID() - changes.get(0).getChangeID() + 1 == size);
+ }
+
+ /**
+ * Given a collection of MSentryChange instances sorted by ID, return the string that prints in the collapsed format.
+ * <p>
+ * e.g:
+ * <li> Input: [MSentryChange(1), MSentryChange(2), MSentryChange(3), MSentryChange(5), MSentryChange(7)] </li>
+ * <li> Output: "[1-3, 5, 7]" </li>
+ * </p>
+ * @param changes List of {@link MSentryChange}
+ * @return Collapsed string representation of the changeIDs
+ */
+ public static String collapseChangeIDsToString(Collection<? extends MSentryChange> changes) {
+ return SentryUtils.collapseNumsToString(getChangeIds(changes));
+ }
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/7db84b2f/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryVersion.java
----------------------------------------------------------------------
diff --git a/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryVersion.java b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryVersion.java
new file mode 100644
index 0000000..b0dbaf0
--- /dev/null
+++ b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/MSentryVersion.java
@@ -0,0 +1,66 @@
+/**
+ * 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.model;
+
+import javax.jdo.annotations.PersistenceCapable;
+
+@PersistenceCapable
+public class MSentryVersion {
+ private String schemaVersion;
+ private String versionComment;
+
+ public MSentryVersion() {
+ }
+
+ public MSentryVersion(String schemaVersion, String versionComment) {
+ this.schemaVersion = schemaVersion.intern();
+ this.versionComment = versionComment.intern();
+ }
+
+ /**
+ * @return the versionComment
+ */
+ public String getVersionComment() {
+ return versionComment;
+ }
+
+ /**
+ * @param versionComment
+ * the versionComment to set
+ */
+ public void setVersionComment(String versionComment) {
+ this.versionComment = versionComment;
+ }
+
+ /**
+ * @return the schemaVersion
+ */
+ public String getSchemaVersion() {
+ return schemaVersion;
+ }
+
+ /**
+ * @param schemaVersion
+ * the schemaVersion to set
+ */
+ public void setSchemaVersion(String schemaVersion) {
+ this.schemaVersion = schemaVersion;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/7db84b2f/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/package.jdo
----------------------------------------------------------------------
diff --git a/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/package.jdo b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/package.jdo
new file mode 100644
index 0000000..6539e33
--- /dev/null
+++ b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/model/package.jdo
@@ -0,0 +1,341 @@
+<?xml version="1.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.
+-->
+<!DOCTYPE jdo PUBLIC "-//Sun Microsystems, Inc.//DTD Java Data Objects Metadata 2.0//EN"
+ "http://java.sun.com/dtd/jdo_2_0.dtd">
+<!--
+ Size Limitations:
+
+ Indexed VARCHAR: 767 bytes (MySQL running on InnoDB Engine http://bugs.mysql.com/bug.php?id=13315)
+ Non-indexed VARCHAR: 4000 bytes (max length on Oracle 9i/10g/11g)
+
+-->
+<jdo>
+ <package name="org.apache.sentry.provider.db.service.model">
+ <class name="MSentryGroup" identity-type="datastore" table="SENTRY_GROUP" detachable="true">
+ <datastore-identity>
+ <column name="GROUP_ID"/>
+ </datastore-identity>
+ <field name="groupName">
+ <column name="GROUP_NAME" length="128" jdbc-type="VARCHAR"/>
+ <index name="SentryGroupName" unique="true"/>
+ </field>
+ <field name = "createTime">
+ <column name = "CREATE_TIME" jdbc-type="BIGINT"/>
+ </field>
+
+ <field name="roles" mapped-by="groups">
+ <collection element-type="org.apache.sentry.provider.db.service.model.MSentryRole"/>
+ </field>
+
+ </class>
+
+ <class name="MSentryUser" identity-type="datastore" table="SENTRY_USER" detachable="true">
+ <datastore-identity>
+ <column name="USER_ID"/>
+ </datastore-identity>
+ <field name="userName">
+ <column name="USER_NAME" length="128" jdbc-type="VARCHAR"/>
+ <index name="SentryUserName" unique="true"/>
+ </field>
+ <field name = "createTime">
+ <column name = "CREATE_TIME" jdbc-type="BIGINT"/>
+ </field>
+
+ <field name="roles" mapped-by="users">
+ <collection element-type="org.apache.sentry.provider.db.service.model.MSentryRole"/>
+ </field>
+
+ <field name = "privileges" table="SENTRY_USER_DB_PRIVILEGE_MAP" default-fetch-group="true">
+ <collection element-type="org.apache.sentry.provider.db.service.model.MSentryPrivilege"/>
+ <join>
+ <column name="USER_ID"/>
+ </join>
+ <element>
+ <column name="DB_PRIVILEGE_ID"/>
+ </element>
+ </field>
+ </class>
+
+ <class name="MSentryRole" identity-type="datastore" table="SENTRY_ROLE" detachable="true">
+ <datastore-identity>
+ <column name="ROLE_ID"/>
+ </datastore-identity>
+ <field name="roleName">
+ <column name="ROLE_NAME" length="128" jdbc-type="VARCHAR"/>
+ <index name="SentryRoleName" unique="true"/>
+ </field>
+ <field name = "createTime">
+ <column name = "CREATE_TIME" jdbc-type="BIGINT"/>
+ </field>
+ <field name = "privileges" table="SENTRY_ROLE_DB_PRIVILEGE_MAP" default-fetch-group="true">
+ <collection element-type="org.apache.sentry.provider.db.service.model.MSentryPrivilege"/>
+ <join>
+ <column name="ROLE_ID"/>
+ </join>
+ <element>
+ <column name="DB_PRIVILEGE_ID"/>
+ </element>
+ </field>
+
+ <field name = "gmPrivileges" table="SENTRY_ROLE_GM_PRIVILEGE_MAP" default-fetch-group="true">
+ <collection element-type="org.apache.sentry.provider.db.service.model.MSentryGMPrivilege"/>
+ <join>
+ <column name="ROLE_ID"/>
+ </join>
+ <element>
+ <column name="GM_PRIVILEGE_ID"/>
+ </element>
+ </field>
+
+ <field name = "groups" table="SENTRY_ROLE_GROUP_MAP" default-fetch-group="true">
+ <collection element-type="org.apache.sentry.provider.db.service.model.MSentryGroup"/>
+ <join>
+ <column name="ROLE_ID"/>
+ </join>
+ <element>
+ <column name="GROUP_ID"/>
+ </element>
+ </field>
+
+ <field name = "users" table="SENTRY_ROLE_USER_MAP" default-fetch-group="true">
+ <collection element-type="org.apache.sentry.provider.db.service.model.MSentryUser"/>
+ <join>
+ <column name="ROLE_ID"/>
+ </join>
+ <element>
+ <column name="USER_ID"/>
+ </element>
+ </field>
+ </class>
+
+ <class name="MSentryPrivilege" identity-type="datastore" table="SENTRY_DB_PRIVILEGE" detachable="true">
+ <datastore-identity>
+ <column name="DB_PRIVILEGE_ID"/>
+ </datastore-identity>
+ <index name="PRIVILEGE_INDEX" unique="true">
+ <field name="serverName"/>
+ <field name="dbName"/>
+ <field name="tableName"/>
+ <field name="columnName"/>
+ <field name="URI"/>
+ <field name="action"/>
+ <field name="grantOption"/>
+ </index>
+ <field name="privilegeScope">
+ <column name="PRIVILEGE_SCOPE" length="40" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="serverName">
+ <column name="SERVER_NAME" length="4000" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="dbName">
+ <column name="DB_NAME" length="4000" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="tableName">
+ <column name="TABLE_NAME" length="4000" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="columnName">
+ <column name="COLUMN_NAME" length="4000" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="URI">
+ <column name="URI" length="4000" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="action">
+ <column name="ACTION" length="40" jdbc-type="VARCHAR"/>
+ </field>
+ <field name = "createTime">
+ <column name = "CREATE_TIME" jdbc-type="BIGINT"/>
+ </field>
+ <field name="grantOption">
+ <column name="WITH_GRANT_OPTION" length="1" jdbc-type="CHAR"/>
+ </field>
+ <field name="roles" mapped-by="privileges">
+ <collection element-type="org.apache.sentry.provider.db.service.model.MSentryRole"/>
+ </field>
+ <field name="users" mapped-by="privileges">
+ <collection element-type="org.apache.sentry.provider.db.service.model.MSentryUser"/>
+ </field>
+ </class>
+
+ <class name="MSentryGMPrivilege" identity-type="datastore" table="SENTRY_GM_PRIVILEGE" detachable="true">
+ <datastore-identity>
+ <column name="GM_PRIVILEGE_ID"/>
+ </datastore-identity>
+ <index name="GM_PRIVILEGE_INDEX" unique="true">
+ <field name="componentName"/>
+ <field name="serviceName"/>
+ <field name="resourceName0"/>
+ <field name="resourceType0"/>
+ <field name="resourceName1"/>
+ <field name="resourceType1"/>
+ <field name="resourceName2"/>
+ <field name="resourceType2"/>
+ <field name="resourceName3"/>
+ <field name="resourceType3"/>
+ <field name="action"/>
+ <field name="grantOption"/>
+ </index>
+ <field name="componentName">
+ <column name="COMPONENT_NAME" length="100" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="serviceName">
+ <column name="SERVICE_NAME" length="100" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="resourceName0">
+ <column name="RESOURCE_NAME_0" length="100" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="resourceType0">
+ <column name="RESOURCE_TYPE_0" length="100" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="resourceName1">
+ <column name="RESOURCE_NAME_1" length="100" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="resourceType1">
+ <column name="RESOURCE_TYPE_1" length="100" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="resourceName2">
+ <column name="RESOURCE_NAME_2" length="100" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="resourceType2">
+ <column name="RESOURCE_TYPE_2" length="100" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="resourceName3">
+ <column name="RESOURCE_NAME_3" length="100" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="resourceType3">
+ <column name="RESOURCE_TYPE_3" length="100" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="action">
+ <column name="ACTION" length="100" jdbc-type="VARCHAR"/>
+ </field>
+ <field name="scope">
+ <column name="SCOPE" length="100" jdbc-type="VARCHAR"/>
+ </field>
+ <field name = "createTime">
+ <column name = "CREATE_TIME" jdbc-type="BIGINT"/>
+ </field>
+ <field name="grantOption">
+ <column name="WITH_GRANT_OPTION" length="1" jdbc-type="CHAR"/>
+ </field>
+ <field name="roles" mapped-by="gmPrivileges">
+ <collection element-type="org.apache.sentry.provider.db.service.model.MSentryRole"/>
+ </field>
+ </class>
+
+ <class name="MSentryVersion" table="SENTRY_VERSION" identity-type="datastore" detachable="true">
+ <datastore-identity>
+ <column name="VER_ID"/>
+ </datastore-identity>
+ <field name ="schemaVersion">
+ <column name="SCHEMA_VERSION" length="127" jdbc-type="VARCHAR" allows-null="false"/>
+ </field>
+ <field name ="versionComment">
+ <column name="VERSION_COMMENT" length="255" jdbc-type="VARCHAR" allows-null="false"/>
+ </field>
+ </class>
+
+ <class name="MAuthzPathsSnapshotId" identity-type="application" table="AUTHZ_PATHS_SNAPSHOT_ID" detachable="true">
+ <field name="authzSnapshotID" primary-key="true">
+ <column name="AUTHZ_SNAPSHOT_ID" jdbc-type="BIGINT" allows-null="false"/>
+ </field>
+ </class>
+
+ <class name="MAuthzPathsMapping" identity-type="datastore" table="AUTHZ_PATHS_MAPPING" detachable="true">
+ <datastore-identity>
+ <column name="AUTHZ_OBJ_ID"/>
+ </datastore-identity>
+ <index name="AUTHZ_SNAPSHOT_ID_INDEX" unique="false">
+ <field name="authzSnapshotID"/>
+ </index>
+ <!--
+ authzObjName is composed by hive database name, and table name. e.g. "default.tb1". Since
+ both hive database name, and table name have restrictions to be at most 128 characters long,
+ 384 characters length should be enough for AUTHZ_OBJ_NAM.
+ -->
+ <field name="authzObjName">
+ <column name="AUTHZ_OBJ_NAME" length="384" jdbc-type="VARCHAR" allows-null="false"/>
+ </field>
+ <field name="createTimeMs">
+ <column name="CREATE_TIME_MS" jdbc-type="BIGINT"/>
+ </field>
+ <field name = "paths">
+ <collection element-type="org.apache.sentry.provider.db.service.model.MPath"/>
+ <element>
+ <column name="AUTHZ_OBJ_ID"/>
+ </element>
+ </field>
+ <fetch-group name="includingPaths">
+ <field name="paths"/>
+ </fetch-group>
+ <field name="authzSnapshotID">
+ <column name="AUTHZ_SNAPSHOT_ID" jdbc-type="BIGINT" allows-null="false"/>
+ </field>
+ </class>
+
+ <class name="MPath" identity-type="datastore" table="AUTHZ_PATH" detachable="true">
+ <datastore-identity>
+ <column name="PATH_ID"/>
+ </datastore-identity>
+ <field name="path">
+ <column name="PATH_NAME" length="4000" jdbc-type="VARCHAR"/>
+ </field>
+ </class>
+
+ <class name="MSentryPermChange" table="SENTRY_PERM_CHANGE" identity-type="application" detachable="true">
+ <field name="changeID" primary-key="true">
+ <column name="CHANGE_ID" jdbc-type="BIGINT" allows-null="false"/>
+ </field>
+ <field name ="permChange">
+ <column name="PERM_CHANGE" length="4000" jdbc-type="VARCHAR" allows-null="false"/>
+ </field>
+ <field name="createTimeMs">
+ <column name="CREATE_TIME_MS" jdbc-type="BIGINT"/>
+ </field>
+ </class>
+
+ <class name="MSentryPathChange" table="SENTRY_PATH_CHANGE" identity-type="application" detachable="true">
+ <field name="changeID" primary-key="true">
+ <column name="CHANGE_ID" jdbc-type="BIGINT" allows-null="false"/>
+ </field>
+ <!--
+ notificationHash is a unique identifier for the HMS notification used to prevent
+ the same HMS notification message to be processed twice.
+ The current HMS code may send different notifications messages with the same ID. To
+ keep this ID unique, we calculate the SHA-1 hash of the full message received.
+ (This is a temporary fix until HIVE-16886 fixes the issue with duplicated IDs)
+ -->
+ <field name="notificationHash">
+ <column name="NOTIFICATION_HASH" jdbc-type="CHAR(40)" allows-null="false"/>
+ <index name="NOTIFICATION_HASH_INDEX" unique="true"/>
+ </field>
+ <field name ="pathChange">
+ <column name="PATH_CHANGE" jdbc-type="LONGVARCHAR" allows-null="false"/>
+ </field>
+ <field name="createTimeMs">
+ <column name="CREATE_TIME_MS" jdbc-type="BIGINT"/>
+ </field>
+ </class>
+
+ <class name="MSentryHmsNotification" table="SENTRY_HMS_NOTIFICATION_ID" identity-type="nondurable" detachable="true">
+ <field name="notificationId">
+ <column name="NOTIFICATION_ID" jdbc-type="BIGINT" allows-null="false"/>
+ <index name="SENTRY_HMS_NOTIF_ID_INDEX"/>
+ </field>
+ </class>
+ </package>
+</jdo>
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/sentry/blob/7db84b2f/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/CounterWait.java
----------------------------------------------------------------------
diff --git a/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/CounterWait.java b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/CounterWait.java
new file mode 100644
index 0000000..d8c8297
--- /dev/null
+++ b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/CounterWait.java
@@ -0,0 +1,341 @@
+/*
+ * 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 org.apache.http.annotation.ThreadSafe;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Waiting for counter to reach certain value.
+ * The counter starts from zero and its value increases over time.
+ * The class allows for multiple consumers waiting until the value of the
+ * counter reaches some value interesting to them.
+ * Consumers call {@link #waitFor(long)} which may either return
+ * immediately if the counter reached the specified value, or block
+ * until this value is reached. Consumers can also specify timeout for the
+ * {@link #waitFor(long)} in which case it may return {@link TimeoutException}
+ * when the wait was not successfull within the specified time limit.
+ * <p>
+ * All waiters should be waken up when the counter becomes equal or higher
+ * then the value they are waiting for.
+ * <p>
+ * The counter is updated by a single updater that should only increase the
+ * counter value.
+ * The updater calls the {@link #update(long)} method to update the counter
+ * value and this should wake up all threads waiting for any value smaller or
+ * equal to the new one.
+ * <p>
+ * The class is thread-safe.
+ * It is designed for use by multiple waiter threads and a single
+ * updater thread, but it will work correctly even in the presence of multiple
+ * updater threads.
+ */
+@ThreadSafe
+public final class CounterWait {
+ // Implementation notes.
+ //
+ // The implementation is based on:
+ //
+ // 1) Using an atomic counter value which guarantees consistency.
+ // Since everyone needs only to know when the counter value reached the
+ // certain value and the counter may only increase its value,
+ // it is safe to update the counter by another thread after its value
+ // was read.
+ //
+ // 2) Priority queue of waiters, sorted by their expected values. The smallest
+ // value is always at the top of the queue. The priority queue itself
+ // is thread-safe, so no locks are needed to protect access to it.
+ //
+ // Each waiter is implemented using a binary semaphore.
+ // This solves the problem of a wakeup that happens before the sleep -
+ // in this case the acquire() doesn't block and returns immediately.
+ //
+ // NOTE: We use PriorityBlockingQueue for waiters because it is thread-safe,
+ // we are not using its blocking queue semantics.
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(CounterWait.class);
+
+ /** Counter value. May only increase. */
+ private final AtomicLong currentId = new AtomicLong(0);
+
+ private final long waitTimeout;
+ private final TimeUnit waitTimeUnit;
+
+ /**
+ * Waiters sorted by the value of the counter they are waiting for.
+ * Note that {@link PriorityBlockingQueue} is thread-safe.
+ * We are not using this as a blocking queue, but as a synchronized
+ * PriorityQueue.
+ */
+ private final PriorityBlockingQueue<ValueEvent> waiters =
+ new PriorityBlockingQueue<>();
+
+ /**
+ * Create an instance of CounterWait object that will not timeout during wait
+ */
+ public CounterWait() {
+ this(0, TimeUnit.SECONDS);
+ }
+
+ /**
+ * Create an instance of CounterWait object that will timeout during wait
+ * @param waitTimeout maximum time in seconds to wait for counter
+ */
+ public CounterWait(long waitTimeoutSec) {
+ this(waitTimeoutSec, TimeUnit.SECONDS);
+ }
+
+ /**
+ * Create an instance of CounterWait object that will timeout during wait
+ * @param waitTimeout maximum time to wait for counter
+ * @param waitTimeUnit time units for wait
+ */
+ public CounterWait(long waitTimeout, TimeUnit waitTimeUnit) {
+ this.waitTimeout = waitTimeout;
+ this.waitTimeUnit = waitTimeUnit;
+ }
+
+ /**
+ * Update the counter value and wake up all threads waiting for this
+ * value or any value below it.
+ * <p>
+ * The counter value should only increase.
+ * An attempt to decrease the value is ignored.
+ *
+ * @param newValue the new counter value
+ */
+ public synchronized void update(long newValue) {
+ // update() is synchronized so the value can't change.
+ long oldValue = currentId.get();
+ LOGGER.debug("CounterWait update: oldValue = {}, newValue = {}", oldValue, newValue);
+ // Avoid doing extra work if not needed
+ if (oldValue == newValue) {
+ return; // no-op
+ }
+
+ // Make sure the counter is never decremented.
+ if (newValue < oldValue) {
+ LOGGER.error("new counter value {} is smaller then the previous one {}",
+ newValue, oldValue);
+ return; // no-op
+ }
+
+ currentId.set(newValue);
+
+ // Wake up any threads waiting for a counter to reach this value.
+ wakeup(newValue);
+ }
+
+ /**
+ * Explicitly reset the counter value to a new value, but allow setting to a
+ * smaller value.
+ * This should be used when we have some external event that resets the counter
+ * value space.
+ * @param newValue New counter value. If this is greater or equal then the current
+ * value, this is equivalent to {@link #update(long)}. Otherwise
+ * sets the counter to the new smaller value.
+ */
+ public synchronized void reset(long newValue) {
+ long oldValue = currentId.get();
+ LOGGER.debug("CounterWait reset: oldValue = {}, newValue = {}", oldValue, newValue);
+
+ if (newValue > oldValue) {
+ update(newValue);
+ } else if (newValue < oldValue) {
+ LOGGER.warn("resetting counter from {} to smaller value {}",
+ oldValue, newValue);
+ currentId.set(newValue);
+ // No need to wakeup waiters since no one should wait on the smaller value
+ }
+ }
+
+
+ /**
+ * Wait for specified counter value.
+ * Returns immediately if the value is reached or blocks until the value
+ * is reached.
+ * Multiple threads can call the method concurrently.
+ *
+ * @param value requested counter value
+ * @return current counter value that should be no smaller then the requested
+ * value
+ * @throws InterruptedException if the wait was interrupted, TimeoutException if
+ * wait was not successfull within the timeout value specified at the construction time.
+ */
+ public long waitFor(long value) throws InterruptedException, TimeoutException {
+ // Fast path - counter value already reached, no need to block
+ if (value <= currentId.get()) {
+ return currentId.get();
+ }
+
+ // Enqueue the waiter for this value
+ ValueEvent eid = new ValueEvent(value);
+ waiters.put(eid);
+
+ // It is possible that between the fast path check and the time the
+ // value event is enqueued, the counter value already reached the requested
+ // value. In this case we return immediately.
+ if (value <= currentId.get()) {
+ return currentId.get();
+ }
+
+ // At this point we may be sure that by the time the event was enqueued,
+ // the counter was below the requested value. This means that update()
+ // is guaranteed to wake us up when the counter reaches the requested value.
+ // The wake up may actually happen before we start waiting, in this case
+ // the event's blocking queue will be non-empty and the waitFor() below
+ // will not block, so it is safe to wake up before the wait.
+ // So sit tight and wait patiently.
+ eid.waitFor();
+ LOGGER.debug("CounterWait added new value to waitFor: value = {}, currentId = {}", value, currentId.get());
+ return currentId.get();
+ }
+
+ /**
+ * Wake up any threads waiting for a counter to reach specified value
+ * Peek at the top of the queue. If the queue is empty or the top value
+ * exceeds the current value, we are done. Otherwise wakeup the top thread,
+ * remove the corresponding waiter and continue.
+ * <p>
+ * Note that the waiter may be removed under our nose by
+ * {@link #waitFor(long)} method, but this is Ok - in this case
+ * waiters.remove() will just return false.
+ *
+ * @param value current counter value
+ */
+ private void wakeup(long value) {
+ while (true) {
+ // Get the top of the waiters queue or null if it is empty
+ ValueEvent e = waiters.poll();
+ if (e == null) {
+ // Queue is empty - return.
+ return;
+ }
+ // No one to wake up, return event to the queue and exit
+ if (e.getValue() > value) {
+ waiters.add(e);
+ return;
+ }
+ // Due for wake-up call
+ LOGGER.debug("CounterWait wakeup: event = {} is less than value = {}", e.getValue(), value);
+ e.wakeup();
+ }
+ }
+
+ // Useful for debugging
+ @Override
+ public String toString() {
+ return "CounterWait{" + "currentId=" + currentId +
+ ", waiters=" + waiters + "}";
+ }
+
+ /**
+ * Return number of waiters. This is mostly useful for metrics/debugging
+ *
+ * @return number of sleeping waiters
+ */
+ public int waitersCount() {
+ return waiters.size();
+ }
+
+ /**
+ * Representation of the waiting event.
+ * The waiting event consists of the expected value and a binary semaphore.
+ * <p>
+ * Each thread waiting for the given value, creates a ValueEvent and tries
+ * to acquire a semaphore. This blocks until the semaphore is released.
+ * <p>
+ * ValueEvents are stored in priority queue sorted by value, so they should be
+ * comparable by the value.
+ */
+ private class ValueEvent implements Comparable<ValueEvent> {
+ /** Value waited for. */
+ private final long value;
+ /** Binary semaphore to synchronize waiters */
+ private final Semaphore semaphore = new Semaphore(1);
+
+ /**
+ * Instantiates a new Value event.
+ *
+ * @param v the expected value
+ */
+ ValueEvent(long v) {
+ this.value = v;
+ // Acquire the semaphore. Subsequent calls to waitFor() will block until
+ // wakeup() releases the semaphore.
+ semaphore.acquireUninterruptibly(); // Will not block
+ }
+
+ /** Wait until signaled or interrupted. May return immediately if already signalled. */
+ void waitFor() throws InterruptedException, TimeoutException {
+ if (waitTimeout == 0) {
+ semaphore.acquire();
+ return;
+ }
+ if (!semaphore.tryAcquire(waitTimeout, waitTimeUnit)) {
+ throw new TimeoutException();
+ }
+ }
+
+ /** @return the value we are waiting for. */
+ long getValue() {
+ return value;
+ }
+
+ /** Wakeup the waiting thread. */
+ void wakeup() {
+ semaphore.release();
+ }
+
+ /**
+ * Compare objects by value.
+ */
+ @Override
+ public int compareTo(final ValueEvent o) {
+ return value == o.value ? 0
+ : value < o.value ? -1
+ : 1;
+ }
+
+ /**
+ * Use identity comparison of objects.
+ */
+ @Override
+ public boolean equals(final Object o) {
+ return (this == o);
+ }
+
+ @Override
+ public int hashCode() {
+ return (int) (value ^ (value >>> 32));
+ }
+
+ @Override
+ public String toString() {
+ return String.valueOf(value);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/7db84b2f/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/DeltaTransactionBlock.java
----------------------------------------------------------------------
diff --git a/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/DeltaTransactionBlock.java b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/DeltaTransactionBlock.java
new file mode 100644
index 0000000..849e5f8
--- /dev/null
+++ b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/DeltaTransactionBlock.java
@@ -0,0 +1,103 @@
+/*
+ * 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.Preconditions;
+import org.apache.sentry.core.common.exception.SentryInvalidInputException;
+import org.apache.sentry.hdfs.PathsUpdate;
+import org.apache.sentry.hdfs.PermissionsUpdate;
+import org.apache.sentry.hdfs.UniquePathsUpdate;
+import org.apache.sentry.provider.db.service.model.MSentryHmsNotification;
+import org.apache.sentry.provider.db.service.model.MSentryPathChange;
+import org.apache.sentry.provider.db.service.model.MSentryPermChange;
+import static org.apache.sentry.hdfs.Updateable.Update;
+
+import javax.jdo.PersistenceManager;
+
+/**
+ * DeltaTransactionBlock is an implementation of {@link TransactionBlock}
+ * that persists delta updates for {@link PathsUpdate} or {@link PermissionsUpdate}
+ * into corresponding update table, e.g {@link MSentryPathChange} or
+ * {@link MSentryPermChange}.
+ * <p>
+ * NullPointerException would be thrown if update is null.
+ * {@link SentryInvalidInputException} would be thrown when update is
+ * neither type of PathsUpdate nor PermissionsUpdate, also in the case
+ * update contains a full image. TException would be thrown if Update
+ * cannot be successfully serialized to JSON string.
+ */
+public class DeltaTransactionBlock implements TransactionBlock<Object> {
+ private final Update update;
+
+ public DeltaTransactionBlock(Update update) {
+ this.update = update;
+ }
+
+ @Override
+ public Object execute(PersistenceManager pm) throws Exception {
+ persistUpdate(pm, update);
+ return null;
+ }
+
+ /**
+ * Persist the delta change into corresponding type based on its type.
+ * Atomic increasing primary key changeID by 1.
+ * <p>
+ * NullPointerException would be thrown if update is null.
+ * {@link SentryInvalidInputException} would be thrown when update is
+ * neither type of PathsUpdate nor PermissionsUpdate. Also in the case
+ * update contains a full image.
+ * TException would be thrown if Update cannot be successfully serialized
+ * to JSON string.
+ *
+ * @param pm PersistenceManager
+ * @param update update
+ * @throws Exception
+ */
+ private void persistUpdate(PersistenceManager pm, Update update)
+ throws Exception {
+ pm.setDetachAllOnCommit(false); // No need to detach objects
+
+ Preconditions.checkNotNull(update);
+ // persistUpdate cannot handle full image update, instead
+ // it only handles delta updates.
+ if (update.hasFullImage()) {
+ throw new SentryInvalidInputException("Update should be not be a full image.\n");
+ }
+
+ // Persist the update into corresponding tables based on its type.
+ // changeID is the primary key in MSentryPXXXChange table. If same
+ // changeID is trying to be persisted twice, the transaction would
+ // fail.
+ if (update instanceof PermissionsUpdate) {
+ long lastChangeID = SentryStore.getLastProcessedChangeIDCore(pm, MSentryPermChange.class);
+ pm.makePersistent(new MSentryPermChange(lastChangeID + 1, (PermissionsUpdate) update));
+ } else if (update instanceof UniquePathsUpdate) {
+ long lastChangeID = SentryStore.getLastProcessedChangeIDCore(pm, MSentryPathChange.class);
+ String eventHash = ((UniquePathsUpdate) update).getEventHash();
+ pm.makePersistent(new MSentryPathChange(lastChangeID + 1, eventHash, (PathsUpdate) update));
+ // Notification id from PATH_UPDATE entry is made persistent in
+ // SENTRY_LAST_NOTIFICATION_ID table.
+ pm.makePersistent(new MSentryHmsNotification(update.getSeqNum()));
+ } else {
+ throw new SentryInvalidInputException("Update should be type of either " +
+ "PermissionsUpdate or PathsUpdate.\n");
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/7db84b2f/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/FixedJsonInstanceSerializer.java
----------------------------------------------------------------------
diff --git a/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/FixedJsonInstanceSerializer.java b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/FixedJsonInstanceSerializer.java
new file mode 100644
index 0000000..476bf6a
--- /dev/null
+++ b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/FixedJsonInstanceSerializer.java
@@ -0,0 +1,163 @@
+/**
+ * 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 java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+import org.codehaus.jackson.JsonNode;
+import org.codehaus.jackson.JsonParseException;
+import org.codehaus.jackson.map.DeserializationConfig;
+import org.codehaus.jackson.map.JsonMappingException;
+import org.codehaus.jackson.map.ObjectMapper;
+
+import com.google.common.base.Preconditions;
+import org.apache.curator.x.discovery.ServiceInstance;
+import org.apache.curator.x.discovery.ServiceInstanceBuilder;
+import org.apache.curator.x.discovery.ServiceType;
+import org.apache.curator.x.discovery.UriSpec;
+import org.apache.curator.x.discovery.details.InstanceSerializer;
+
+// TODO: Workaround for CURATOR-5 (https://issues.apache.org/jira/browse/CURATOR-5)
+// Remove this class (code from pull request listed on JIRA) and use regular JsonInstanceSerializer once fixed
+// (Otherwise we can't properly serialize objects for the ZK Service Discovery)
+public class FixedJsonInstanceSerializer<T> implements InstanceSerializer<T>
+{
+
+ private final ObjectMapper mMapper;
+ private final Class<T> mPayloadClass;
+
+ /**
+ * @param payloadClass
+ * used to validate payloads when deserializing
+ */
+ public FixedJsonInstanceSerializer(final Class<T> payloadClass) {
+ this(payloadClass, new ObjectMapper());
+ }
+
+ public FixedJsonInstanceSerializer(final Class<T> pPayloadClass, final ObjectMapper pMapper) {
+ mPayloadClass = pPayloadClass;
+ mMapper = pMapper;
+ mMapper.configure(DeserializationConfig.Feature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ }
+
+ @Override
+ public byte[] serialize(final ServiceInstance<T> pInstance) throws Exception {
+ final ByteArrayOutputStream out = new ByteArrayOutputStream();
+ mMapper.writeValue(out, pInstance);
+ return out.toByteArray();
+
+ }
+
+ private String getTextField(final JsonNode pNode, final String pFieldName) {
+ Preconditions.checkNotNull(pNode);
+ Preconditions.checkNotNull(pFieldName);
+ return pNode.get(pFieldName) != null ? pNode.get(pFieldName).getTextValue() : null;
+ }
+
+ private Integer getIntegerField(final JsonNode pNode, final String pFieldName) {
+ Preconditions.checkNotNull(pNode);
+ Preconditions.checkNotNull(pFieldName);
+ return pNode.get(pFieldName) != null && pNode.get(pFieldName).isNumber() ? pNode.get(pFieldName)
+ .getIntValue() : null;
+ }
+
+ private Long getLongField(final JsonNode pNode, final String pFieldName) {
+ Preconditions.checkNotNull(pNode);
+ Preconditions.checkNotNull(pFieldName);
+ return pNode.get(pFieldName) != null && pNode.get(pFieldName).isLong() ? pNode.get(pFieldName).getLongValue()
+ : null;
+ }
+
+ private <O> O getObject(final JsonNode pNode, final String pFieldName, final Class<O> pObjectClass)
+ throws JsonParseException, JsonMappingException, IOException {
+ Preconditions.checkNotNull(pNode);
+ Preconditions.checkNotNull(pFieldName);
+ Preconditions.checkNotNull(pObjectClass);
+ if (pNode.get(pFieldName) != null && pNode.get(pFieldName).isObject()) {
+ return mMapper.readValue(pNode.get(pFieldName), pObjectClass);
+ } else {
+ return null;
+ }
+ }
+
+ @Override
+ public ServiceInstance<T> deserialize(final byte[] pBytes) throws Exception {
+ final ByteArrayInputStream bais = new ByteArrayInputStream(pBytes);
+ final JsonNode rootNode = mMapper.readTree(bais);
+ final ServiceInstanceBuilder<T> builder = ServiceInstance.builder();
+ {
+ final String address = getTextField(rootNode, "address");
+ if (address != null) {
+ builder.address(address);
+ }
+ }
+ {
+ final String id = getTextField(rootNode, "id");
+ if (id != null) {
+ builder.id(id);
+ }
+ }
+ {
+ final String name = getTextField(rootNode, "name");
+ if (name != null) {
+ builder.name(name);
+ }
+ }
+ {
+ final Integer port = getIntegerField(rootNode, "port");
+ if (port != null) {
+ builder.port(port);
+ }
+ }
+ {
+ final Integer sslPort = getIntegerField(rootNode, "sslPort");
+ if (sslPort != null) {
+ builder.sslPort(sslPort);
+ }
+ }
+ {
+ final Long registrationTimeUTC = getLongField(rootNode, "registrationTimeUTC");
+ if (registrationTimeUTC != null) {
+ builder.registrationTimeUTC(registrationTimeUTC);
+ }
+ }
+ {
+ final T payload = getObject(rootNode, "payload", mPayloadClass);
+ if (payload != null) {
+ builder.payload(payload);
+ }
+ }
+ {
+ final ServiceType serviceType = getObject(rootNode, "serviceType", ServiceType.class);
+ if (serviceType != null) {
+ builder.serviceType(serviceType);
+ }
+ }
+ {
+ final UriSpec uriSpec = getObject(rootNode, "uriSpec", UriSpec.class);
+ if (uriSpec != null) {
+ builder.uriSpec(uriSpec);
+ }
+ }
+ return builder.build();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/sentry/blob/7db84b2f/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/HAContext.java
----------------------------------------------------------------------
diff --git a/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/HAContext.java b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/HAContext.java
new file mode 100644
index 0000000..2505da9
--- /dev/null
+++ b/sentry-service/sentry-service-server/src/main/java/org/apache/sentry/provider/db/service/persistent/HAContext.java
@@ -0,0 +1,300 @@
+/*
+ * 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 com.google.common.util.concurrent.ThreadFactoryBuilder;
+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 java.util.concurrent.ThreadFactory;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.sentry.service.common.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 static final String SHUTDOWN_THREAD_NAME = "ha-context-shutdown";
+ 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);
+ String ns = conf.get(SENTRY_HA_ZOOKEEPER_NAMESPACE, SENTRY_HA_ZOOKEEPER_NAMESPACE_DEFAULT);
+ // Namespace shouldn't start with slash.
+ // If config namespace starts with slash, remove it first
+ this.namespace = ns.startsWith("/") ? ns.substring(1) : ns;
+
+ 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();
+ ThreadFactory haContextShutdownThreadFactory = new ThreadFactoryBuilder()
+ .setDaemon(false)
+ .setNameFormat(SHUTDOWN_THREAD_NAME)
+ .build();
+ Runtime.getRuntime()
+ .addShutdownHook(haContextShutdownThreadFactory
+ .newThread(new Runnable() {
+ @Override
+ public void run() {
+ LOGGER.info("ShutdownHook closing curator framework");
+ try {
+ if (serverHAContext != null) {
+ 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;
+ }
+
+ /**
+ * Reset existing HA context.
+ * Should be only used by tests to provide different configurations.
+ */
+ public static void resetHAContext() {
+ HAContext oldContext = serverHAContext;
+ if (oldContext != null) {
+ try {
+ oldContext.close();
+ } catch (Exception e) {
+ LOGGER.error("Failed to close HACOntext", e);
+ }
+ }
+ serverHAContext = null;
+ }
+
+
+ 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) {
+ 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;
+ }
+ }
+}