You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@sentry.apache.org by ls...@apache.org on 2014/11/13 08:43:40 UTC
[3/9] incubator-sentry git commit: SENTRY-432: Synchronization of
HDFS permissions to Sentry permissions (Arun Suresh via Lenni Kuff)
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryPermissions.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryPermissions.java b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryPermissions.java
new file mode 100644
index 0000000..4b27e7b
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryPermissions.java
@@ -0,0 +1,220 @@
+/**
+ * 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.hdfs;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.AclEntryScope;
+import org.apache.hadoop.fs.permission.AclEntryType;
+import org.apache.hadoop.fs.permission.FsAction;
+
+import com.google.common.collect.Lists;
+
+public class SentryPermissions implements AuthzPermissions {
+
+ public static class PrivilegeInfo {
+ private final String authzObj;
+ private final Map<String, FsAction> roleToPermission = new HashMap<String, FsAction>();
+ public PrivilegeInfo(String authzObj) {
+ this.authzObj = authzObj;
+ }
+ public PrivilegeInfo setPermission(String role, FsAction perm) {
+ roleToPermission.put(role, perm);
+ return this;
+ }
+ public PrivilegeInfo removePermission(String role) {
+ roleToPermission.remove(role);
+ return this;
+ }
+ public FsAction getPermission(String role) {
+ return roleToPermission.get(role);
+ }
+ public Map<String, FsAction> getAllPermissions() {
+ return roleToPermission;
+ }
+ public String getAuthzObj() {
+ return authzObj;
+ }
+ }
+
+ public static class RoleInfo {
+ private final String role;
+ private final Set<String> groups = new HashSet<String>();
+ public RoleInfo(String role) {
+ this.role = role;
+ }
+ public RoleInfo addGroup(String group) {
+ groups.add(group);
+ return this;
+ }
+ public RoleInfo delGroup(String group) {
+ groups.remove(group);
+ return this;
+ }
+ public String getRole() {
+ return role;
+ }
+ public Set<String> getAllGroups() {
+ return groups;
+ }
+ }
+
+ private final Map<String, PrivilegeInfo> privileges = new HashMap<String, PrivilegeInfo>();
+ private final Map<String, RoleInfo> roles = new HashMap<String, RoleInfo>();
+ private Map<String, Set<String>> authzObjChildren = new HashMap<String, Set<String>>();
+
+ String getParentAuthzObject(String authzObject) {
+ int dot = authzObject.indexOf('.');
+ if (dot > 0) {
+ return authzObject.substring(0, dot);
+ } else {
+ return authzObject;
+ }
+ }
+
+ void addParentChildMappings(String authzObject) {
+ String parent = getParentAuthzObject(authzObject);
+ if (parent != null) {
+ Set<String> children = authzObjChildren.get(parent);
+ if (children == null) {
+ children = new HashSet<String>();
+ authzObjChildren.put(parent, children);
+ }
+ children.add(authzObject);
+ }
+ }
+
+ void removeParentChildMappings(String authzObject) {
+ String parent = getParentAuthzObject(authzObject);
+ if (parent != null) {
+ Set<String> children = authzObjChildren.get(parent);
+ if (children != null) {
+ children.remove(authzObject);
+ }
+ } else {
+ // is parent
+ authzObjChildren.remove(authzObject);
+ }
+ }
+
+ private Map<String, FsAction> getGroupPerms(String authzObj) {
+ Map<String, FsAction> groupPerms = new HashMap<String, FsAction>();
+ if (authzObj == null) {
+ return groupPerms;
+ }
+ PrivilegeInfo privilegeInfo = privileges.get(authzObj);
+ if (privilegeInfo != null) {
+ for (Map.Entry<String, FsAction> privs : privilegeInfo
+ .getAllPermissions().entrySet()) {
+ constructAclEntry(privs.getKey(), privs.getValue(), groupPerms);
+ }
+ }
+ return groupPerms;
+ }
+
+ @Override
+ public List<AclEntry> getAcls(String authzObj) {
+ Map<String, FsAction> groupPerms = getGroupPerms(authzObj);
+ String parent = getParentAuthzObject(authzObj);
+ Map<String, FsAction> pGroupPerms = null;
+ if (parent == null) {
+ pGroupPerms = new HashMap<String, FsAction>();
+ } else {
+ pGroupPerms = getGroupPerms(getParentAuthzObject(authzObj));
+ if ((groupPerms == null)||(groupPerms.size() == 0)) {
+ groupPerms = pGroupPerms;
+ }
+ }
+ List<AclEntry> retList = new LinkedList<AclEntry>();
+ for (Map.Entry<String, FsAction> groupPerm : groupPerms.entrySet()) {
+ AclEntry.Builder builder = new AclEntry.Builder();
+ builder.setName(groupPerm.getKey());
+ builder.setType(AclEntryType.GROUP);
+ builder.setScope(AclEntryScope.ACCESS);
+ FsAction action = groupPerm.getValue();
+ FsAction pAction = pGroupPerms.get(groupPerm.getKey());
+ if (pAction != null) {
+ action.or(pAction);
+ }
+ if ((action == FsAction.READ) || (action == FsAction.WRITE)
+ || (action == FsAction.READ_WRITE)) {
+ action = action.or(FsAction.EXECUTE);
+ }
+ builder.setPermission(action);
+ retList.add(builder.build());
+ }
+ return retList;
+ }
+
+ private void constructAclEntry(String role, FsAction permission,
+ Map<String, FsAction> groupPerms) {
+ RoleInfo roleInfo = roles.get(role);
+ if (roleInfo != null) {
+ for (String group : roleInfo.groups) {
+ FsAction fsAction = groupPerms.get(group);
+ if (fsAction == null) {
+ fsAction = FsAction.NONE;
+ }
+ groupPerms.put(group, fsAction.or(permission));
+ }
+ }
+ }
+
+ public PrivilegeInfo getPrivilegeInfo(String authzObj) {
+ return privileges.get(authzObj);
+ }
+
+ Collection<PrivilegeInfo> getAllPrivileges() {
+ return privileges.values();
+ }
+
+ Collection<RoleInfo> getAllRoles() {
+ return roles.values();
+ }
+
+ public void delPrivilegeInfo(String authzObj) {
+ privileges.remove(authzObj);
+ }
+
+ public void addPrivilegeInfo(PrivilegeInfo privilegeInfo) {
+ privileges.put(privilegeInfo.authzObj, privilegeInfo);
+ }
+
+ public Set<String> getChildren(String authzObj) {
+ return authzObjChildren.get(authzObj);
+ }
+
+ public RoleInfo getRoleInfo(String role) {
+ return roles.get(role);
+ }
+
+ public void delRoleInfo(String role) {
+ roles.remove(role);
+ }
+
+ public void addRoleInfo(RoleInfo roleInfo) {
+ roles.put(roleInfo.role, roleInfo);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryUpdater.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryUpdater.java b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryUpdater.java
new file mode 100644
index 0000000..9540397
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/SentryUpdater.java
@@ -0,0 +1,61 @@
+/**
+ * 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.hdfs;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.sentry.hdfs.SentryHDFSServiceClient.SentryAuthzUpdate;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SentryUpdater {
+
+ private SentryHDFSServiceClient sentryClient;
+ private final Configuration conf;
+ private final SentryAuthorizationInfo authzInfo;
+
+ private static Logger LOG = LoggerFactory.getLogger(SentryUpdater.class);
+
+ public SentryUpdater(Configuration conf, SentryAuthorizationInfo authzInfo) throws Exception {
+ this.conf = conf;
+ this.authzInfo = authzInfo;
+ }
+
+ public SentryAuthzUpdate getUpdates() {
+ if (sentryClient == null) {
+ try {
+ sentryClient = new SentryHDFSServiceClient(conf);
+ } catch (Exception e) {
+ LOG.error("Error connecting to Sentry ['{}'] !!",
+ e.getMessage());
+ sentryClient = null;
+ return null;
+ }
+ }
+ try {
+ SentryAuthzUpdate sentryUpdates = sentryClient.getAllUpdatesFrom(
+ authzInfo.getAuthzPermissions().getLastUpdatedSeqNum() + 1,
+ authzInfo.getAuthzPaths().getLastUpdatedSeqNum() + 1);
+ return sentryUpdates;
+ } catch (Exception e) {
+ sentryClient = null;
+ LOG.error("Error receiving updates from Sentry !!", e);
+ return null;
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPermissions.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPermissions.java b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPermissions.java
new file mode 100644
index 0000000..e5af802
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/main/java/org/apache/sentry/hdfs/UpdateableAuthzPermissions.java
@@ -0,0 +1,230 @@
+/**
+ * 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.hdfs;
+
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.sentry.hdfs.SentryPermissions.PrivilegeInfo;
+import org.apache.sentry.hdfs.SentryPermissions.RoleInfo;
+import org.apache.sentry.hdfs.service.thrift.TPrivilegeChanges;
+import org.apache.sentry.hdfs.service.thrift.TRoleChanges;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class UpdateableAuthzPermissions implements AuthzPermissions, Updateable<PermissionsUpdate> {
+ private static final int MAX_UPDATES_PER_LOCK_USE = 99;
+ private volatile SentryPermissions perms = new SentryPermissions();
+ private final AtomicLong seqNum = new AtomicLong(0);
+
+ private static Logger LOG = LoggerFactory.getLogger(UpdateableAuthzPermissions.class);
+
+ public static Map<String, FsAction> ACTION_MAPPING = new HashMap<String, FsAction>();
+ static {
+ ACTION_MAPPING.put("ALL", FsAction.ALL);
+ ACTION_MAPPING.put("*", FsAction.ALL);
+ ACTION_MAPPING.put("SELECT", FsAction.READ_EXECUTE);
+ ACTION_MAPPING.put("select", FsAction.READ_EXECUTE);
+ ACTION_MAPPING.put("INSERT", FsAction.WRITE_EXECUTE);
+ ACTION_MAPPING.put("insert", FsAction.WRITE_EXECUTE);
+ }
+
+ @Override
+ public List<AclEntry> getAcls(String authzObj) {
+ return perms.getAcls(authzObj);
+ }
+
+ @Override
+ public UpdateableAuthzPermissions updateFull(PermissionsUpdate update) {
+ UpdateableAuthzPermissions other = new UpdateableAuthzPermissions();
+ other.applyPartialUpdate(update);
+ other.seqNum.set(update.getSeqNum());
+ return other;
+ }
+
+ @Override
+ public void updatePartial(Iterable<PermissionsUpdate> updates, ReadWriteLock lock) {
+ lock.writeLock().lock();
+ try {
+ int counter = 0;
+ for (PermissionsUpdate update : updates) {
+ applyPartialUpdate(update);
+ if (++counter > MAX_UPDATES_PER_LOCK_USE) {
+ counter = 0;
+ lock.writeLock().unlock();
+ lock.writeLock().lock();
+ }
+ seqNum.set(update.getSeqNum());
+ LOG.debug("##### Updated perms seq Num [" + seqNum.get() + "]");
+ }
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+
+ private void applyPartialUpdate(PermissionsUpdate update) {
+ applyPrivilegeUpdates(update);
+ applyRoleUpdates(update);
+ }
+
+ private void applyRoleUpdates(PermissionsUpdate update) {
+ for (TRoleChanges rUpdate : update.getRoleUpdates()) {
+ if (rUpdate.getRole().equals(PermissionsUpdate.ALL_ROLES)) {
+ // Request to remove group from all roles
+ String groupToRemove = rUpdate.getDelGroups().iterator().next();
+ for (RoleInfo rInfo : perms.getAllRoles()) {
+ rInfo.delGroup(groupToRemove);
+ }
+ }
+ RoleInfo rInfo = perms.getRoleInfo(rUpdate.getRole());
+ for (String group : rUpdate.getAddGroups()) {
+ if (rInfo == null) {
+ rInfo = new RoleInfo(rUpdate.getRole());
+ }
+ rInfo.addGroup(group);
+ }
+ if (rInfo != null) {
+ perms.addRoleInfo(rInfo);
+ for (String group : rUpdate.getDelGroups()) {
+ if (group.equals(PermissionsUpdate.ALL_GROUPS)) {
+ perms.delRoleInfo(rInfo.getRole());
+ break;
+ }
+ // If there are no groups to remove, rUpdate.getDelGroups() will
+ // return empty list and this code will not be reached
+ rInfo.delGroup(group);
+ }
+ }
+ }
+ }
+
+ private void applyPrivilegeUpdates(PermissionsUpdate update) {
+ for (TPrivilegeChanges pUpdate : update.getPrivilegeUpdates()) {
+ if (pUpdate.getAuthzObj().equals(PermissionsUpdate.RENAME_PRIVS)) {
+ String newAuthzObj = pUpdate.getAddPrivileges().keySet().iterator().next();
+ String oldAuthzObj = pUpdate.getDelPrivileges().keySet().iterator().next();
+ PrivilegeInfo privilegeInfo = perms.getPrivilegeInfo(oldAuthzObj);
+ Map<String, FsAction> allPermissions = privilegeInfo.getAllPermissions();
+ perms.delPrivilegeInfo(oldAuthzObj);
+ perms.removeParentChildMappings(oldAuthzObj);
+ PrivilegeInfo newPrivilegeInfo = new PrivilegeInfo(newAuthzObj);
+ for (Map.Entry<String, FsAction> e : allPermissions.entrySet()) {
+ newPrivilegeInfo.setPermission(e.getKey(), e.getValue());
+ }
+ perms.addPrivilegeInfo(newPrivilegeInfo);
+ perms.addParentChildMappings(newAuthzObj);
+ return;
+ }
+ if (pUpdate.getAuthzObj().equals(PermissionsUpdate.ALL_AUTHZ_OBJ)) {
+ // Request to remove role from all Privileges
+ String roleToRemove = pUpdate.getDelPrivileges().keySet().iterator()
+ .next();
+ for (PrivilegeInfo pInfo : perms.getAllPrivileges()) {
+ pInfo.removePermission(roleToRemove);
+ }
+ }
+ PrivilegeInfo pInfo = perms.getPrivilegeInfo(pUpdate.getAuthzObj());
+ for (Map.Entry<String, String> aMap : pUpdate.getAddPrivileges().entrySet()) {
+ if (pInfo == null) {
+ pInfo = new PrivilegeInfo(pUpdate.getAuthzObj());
+ }
+ FsAction fsAction = pInfo.getPermission(aMap.getKey());
+ if (fsAction == null) {
+ fsAction = getFAction(aMap.getValue());
+ } else {
+ fsAction = fsAction.or(getFAction(aMap.getValue()));
+ }
+ pInfo.setPermission(aMap.getKey(), fsAction);
+ }
+ if (pInfo != null) {
+ perms.addPrivilegeInfo(pInfo);
+ perms.addParentChildMappings(pUpdate.getAuthzObj());
+ for (Map.Entry<String, String> dMap : pUpdate.getDelPrivileges().entrySet()) {
+ if (dMap.getKey().equals(PermissionsUpdate.ALL_ROLES)) {
+ // Remove all privileges
+ perms.delPrivilegeInfo(pUpdate.getAuthzObj());
+ perms.removeParentChildMappings(pUpdate.getAuthzObj());
+ break;
+ }
+ List<PrivilegeInfo> parentAndChild = new LinkedList<PrivilegeInfo>();
+ parentAndChild.add(pInfo);
+ Set<String> children = perms.getChildren(pInfo.getAuthzObj());
+ if (children != null) {
+ for (String child : children) {
+ parentAndChild.add(perms.getPrivilegeInfo(child));
+ }
+ }
+ // recursive revoke
+ for (PrivilegeInfo pInfo2 : parentAndChild) {
+ FsAction fsAction = pInfo2.getPermission(dMap.getKey());
+ if (fsAction != null) {
+ fsAction = fsAction.and(getFAction(dMap.getValue()).not());
+ if (FsAction.NONE == fsAction) {
+ pInfo2.removePermission(dMap.getKey());
+ } else {
+ pInfo2.setPermission(dMap.getKey(), fsAction);
+ }
+ }
+ }
+ }
+ }
+ }
+ }
+
+ static FsAction getFAction(String sentryPriv) {
+ String[] strPrivs = sentryPriv.trim().split(",");
+ FsAction retVal = FsAction.NONE;
+ for (String strPriv : strPrivs) {
+ retVal = retVal.or(ACTION_MAPPING.get(strPriv.toUpperCase()));
+ }
+ return retVal;
+ }
+
+ @Override
+ public long getLastUpdatedSeqNum() {
+ return seqNum.get();
+ }
+
+ @Override
+ public PermissionsUpdate createFullImageUpdate(long currSeqNum) {
+ PermissionsUpdate retVal = new PermissionsUpdate(currSeqNum, true);
+ for (PrivilegeInfo pInfo : perms.getAllPrivileges()) {
+ TPrivilegeChanges pUpdate = retVal.addPrivilegeUpdate(pInfo.getAuthzObj());
+ for (Map.Entry<String, FsAction> ent : pInfo.getAllPermissions().entrySet()) {
+ pUpdate.putToAddPrivileges(ent.getKey(), ent.getValue().SYMBOL);
+ }
+ }
+ for (RoleInfo rInfo : perms.getAllRoles()) {
+ TRoleChanges rUpdate = retVal.addRoleUpdate(rInfo.getRole());
+ for (String group : rInfo.getAllGroups()) {
+ rUpdate.addToAddGroups(group);
+ }
+ }
+ return retVal;
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/java/org/apache/sentry/hdfs/MockSentryAuthorizationProvider.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/java/org/apache/sentry/hdfs/MockSentryAuthorizationProvider.java b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/java/org/apache/sentry/hdfs/MockSentryAuthorizationProvider.java
new file mode 100644
index 0000000..2085b52
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/java/org/apache/sentry/hdfs/MockSentryAuthorizationProvider.java
@@ -0,0 +1,26 @@
+/**
+ * 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.hdfs;
+
+public class MockSentryAuthorizationProvider extends
+ SentryAuthorizationProvider {
+
+ public MockSentryAuthorizationProvider() {
+ super(new SentryAuthorizationInfoX());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/java/org/apache/sentry/hdfs/SentryAuthorizationInfoX.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/java/org/apache/sentry/hdfs/SentryAuthorizationInfoX.java b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/java/org/apache/sentry/hdfs/SentryAuthorizationInfoX.java
new file mode 100644
index 0000000..7a1539b
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/java/org/apache/sentry/hdfs/SentryAuthorizationInfoX.java
@@ -0,0 +1,85 @@
+/**
+ * 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.hdfs;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.AclEntryScope;
+import org.apache.hadoop.fs.permission.AclEntryType;
+import org.apache.hadoop.fs.permission.FsAction;
+
+public class SentryAuthorizationInfoX extends SentryAuthorizationInfo {
+
+ public SentryAuthorizationInfoX() {
+ super();
+ }
+
+ @Override
+ public void run() {
+
+ }
+
+ @Override
+ public void start() {
+
+ }
+
+ @Override
+ public void stop() {
+
+ }
+
+ @Override
+ public boolean isStale() {
+ return false;
+ }
+
+ private static final String[] MANAGED = {"user", "authz"};
+ private static final String[] AUTHZ_OBJ = {"user", "authz", "obj"};
+
+ private boolean hasPrefix(String[] prefix, String[] pathElement) {
+ int i = 0;
+ for (; i < prefix.length && i < pathElement.length; i ++) {
+ if (!prefix[i].equals(pathElement[i])) {
+ return false;
+ }
+ }
+ return (i == prefix.length);
+ }
+
+ @Override
+ public boolean isManaged(String[] pathElements) {
+ return hasPrefix(MANAGED, pathElements);
+ }
+
+ @Override
+ public boolean doesBelongToAuthzObject(String[] pathElements) {
+ return hasPrefix(AUTHZ_OBJ, pathElements);
+ }
+
+ @Override
+ public List<AclEntry> getAclEntries(String[] pathElements) {
+ AclEntry acl = new AclEntry.Builder().setType(AclEntryType.USER).
+ setPermission(FsAction.ALL).setName("user-authz").
+ setScope(AclEntryScope.ACCESS).build();
+ return Arrays.asList(acl);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/java/org/apache/sentry/hdfs/TestSentryAuthorizationProvider.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/java/org/apache/sentry/hdfs/TestSentryAuthorizationProvider.java b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/java/org/apache/sentry/hdfs/TestSentryAuthorizationProvider.java
new file mode 100644
index 0000000..b766a8f
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/java/org/apache/sentry/hdfs/TestSentryAuthorizationProvider.java
@@ -0,0 +1,164 @@
+/**
+ * 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.hdfs;
+
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.AclEntryScope;
+import org.apache.hadoop.fs.permission.AclEntryType;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.hdfs.DFSConfigKeys;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.server.namenode.EditLogFileOutputStream;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+
+public class TestSentryAuthorizationProvider {
+ private MiniDFSCluster miniDFS;
+ private UserGroupInformation admin;
+
+ @Before
+ public void setUp() throws Exception {
+ admin = UserGroupInformation.createUserForTesting(
+ System.getProperty("user.name"), new String[] { "supergroup" });
+ admin.doAs(new PrivilegedExceptionAction<Void>() {
+ @Override
+ public Void run() throws Exception {
+ System.setProperty(MiniDFSCluster.PROP_TEST_BUILD_DATA, "target/test/data");
+ Configuration conf = new HdfsConfiguration();
+ conf.setBoolean("sentry.authorization-provider.include-hdfs-authz-as-acl", true);
+ conf.set(DFSConfigKeys.DFS_NAMENODE_AUTHORIZATION_PROVIDER_KEY,
+ MockSentryAuthorizationProvider.class.getName());
+ conf.setBoolean(DFSConfigKeys.DFS_NAMENODE_ACLS_ENABLED_KEY, true);
+ EditLogFileOutputStream.setShouldSkipFsyncForTesting(true);
+ miniDFS = new MiniDFSCluster.Builder(conf).build();
+ return null;
+ }
+ });
+ }
+
+ @After
+ public void cleanUp() throws IOException {
+ if (miniDFS != null) {
+ miniDFS.shutdown();
+ }
+ }
+
+ @Test
+ public void testProvider() throws Exception {
+ admin.doAs(new PrivilegedExceptionAction<Void>() {
+ @Override
+ public Void run() throws Exception {
+ String sysUser = UserGroupInformation.getCurrentUser().getShortUserName();
+ FileSystem fs = FileSystem.get(miniDFS.getConfiguration(0));
+
+ List<AclEntry> baseAclList = new ArrayList<AclEntry>();
+ AclEntry.Builder builder = new AclEntry.Builder();
+ baseAclList.add(builder.setType(AclEntryType.USER)
+ .setScope(AclEntryScope.ACCESS).build());
+ baseAclList.add(builder.setType(AclEntryType.GROUP)
+ .setScope(AclEntryScope.ACCESS).build());
+ baseAclList.add(builder.setType(AclEntryType.OTHER)
+ .setScope(AclEntryScope.ACCESS).build());
+ Path path1 = new Path("/user/authz/obj/xxx");
+ fs.mkdirs(path1);
+ fs.setAcl(path1, baseAclList);
+
+ fs.mkdirs(new Path("/user/authz/xxx"));
+ fs.mkdirs(new Path("/user/xxx"));
+
+ // root
+ Path path = new Path("/");
+ Assert.assertEquals(sysUser, fs.getFileStatus(path).getOwner());
+ Assert.assertEquals("supergroup", fs.getFileStatus(path).getGroup());
+ Assert.assertEquals(new FsPermission((short) 0755), fs.getFileStatus(path).getPermission());
+ Assert.assertTrue(fs.getAclStatus(path).getEntries().isEmpty());
+
+ // dir before prefixes
+ path = new Path("/user");
+ Assert.assertEquals(sysUser, fs.getFileStatus(path).getOwner());
+ Assert.assertEquals("supergroup", fs.getFileStatus(path).getGroup());
+ Assert.assertEquals(new FsPermission((short) 0755), fs.getFileStatus(path).getPermission());
+ Assert.assertTrue(fs.getAclStatus(path).getEntries().isEmpty());
+
+ // prefix dir
+ path = new Path("/user/authz");
+ Assert.assertEquals(sysUser, fs.getFileStatus(path).getOwner());
+ Assert.assertEquals("supergroup", fs.getFileStatus(path).getGroup());
+ Assert.assertEquals(new FsPermission((short) 0755), fs.getFileStatus(path).getPermission());
+ Assert.assertTrue(fs.getAclStatus(path).getEntries().isEmpty());
+
+ // dir inside of prefix, no obj
+ path = new Path("/user/authz/xxx");
+ FileStatus status = fs.getFileStatus(path);
+ Assert.assertEquals(sysUser, status.getOwner());
+ Assert.assertEquals("supergroup", status.getGroup());
+ Assert.assertEquals(new FsPermission((short) 0755), status.getPermission());
+ Assert.assertTrue(fs.getAclStatus(path).getEntries().isEmpty());
+
+ // dir inside of prefix, obj
+ path = new Path("/user/authz/obj");
+ Assert.assertEquals("hive", fs.getFileStatus(path).getOwner());
+ Assert.assertEquals("hive", fs.getFileStatus(path).getGroup());
+ Assert.assertEquals(new FsPermission((short) 0770), fs.getFileStatus(path).getPermission());
+ Assert.assertFalse(fs.getAclStatus(path).getEntries().isEmpty());
+
+ List<AclEntry> acls = new ArrayList<AclEntry>();
+ acls.add(new AclEntry.Builder().setName(sysUser).setType(AclEntryType.USER).setScope(AclEntryScope.ACCESS).setPermission(FsAction.ALL).build());
+ acls.add(new AclEntry.Builder().setName("supergroup").setType(AclEntryType.GROUP).setScope(AclEntryScope.ACCESS).setPermission(FsAction.READ_EXECUTE).build());
+ acls.add(new AclEntry.Builder().setName(null).setType(AclEntryType.OTHER).setScope(AclEntryScope.ACCESS).setPermission(FsAction.READ_EXECUTE).build());
+ acls.add(new AclEntry.Builder().setName("user-authz").setType(AclEntryType.USER).setScope(AclEntryScope.ACCESS).setPermission(FsAction.ALL).build());
+ Assert.assertEquals(new LinkedHashSet<AclEntry>(acls), new LinkedHashSet<AclEntry>(fs.getAclStatus(path).getEntries()));
+
+ // dir inside of prefix, inside of obj
+ path = new Path("/user/authz/obj/xxx");
+ Assert.assertEquals("hive", fs.getFileStatus(path).getOwner());
+ Assert.assertEquals("hive", fs.getFileStatus(path).getGroup());
+ Assert.assertEquals(new FsPermission((short) 0770), fs.getFileStatus(path).getPermission());
+ Assert.assertFalse(fs.getAclStatus(path).getEntries().isEmpty());
+
+ Path path2 = new Path("/user/authz/obj/path2");
+ fs.mkdirs(path2);
+ fs.setAcl(path2, baseAclList);
+
+ // dir outside of prefix
+ path = new Path("/user/xxx");
+ Assert.assertEquals(sysUser, fs.getFileStatus(path).getOwner());
+ Assert.assertEquals("supergroup", fs.getFileStatus(path).getGroup());
+ Assert.assertEquals(new FsPermission((short) 0755), fs.getFileStatus(path).getPermission());
+ Assert.assertTrue(fs.getAclStatus(path).getEntries().isEmpty());
+ return null;
+ }
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/resources/hdfs-sentry.xml
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/resources/hdfs-sentry.xml b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/resources/hdfs-sentry.xml
new file mode 100644
index 0000000..511bfdd
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-namenode-plugin/src/test/resources/hdfs-sentry.xml
@@ -0,0 +1,33 @@
+<?xml version="1.0"?>
+<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
+<!--
+ 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.
+-->
+
+<configuration>
+ <property>
+ <name>sentry.hdfs-plugin.path-prefixes</name>
+ <value>/user/hive/dw</value>
+ </property>
+ <property>
+ <name>sentry.hdfs-plugin.sentry-uri</name>
+ <value>thrift://localhost:1234</value>
+ </property>
+ <property>
+ <name>sentry.hdfs-plugin.stale-threshold.ms</name>
+ <value>-1</value>
+ </property>
+</configuration>
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-service/.gitignore
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/.gitignore b/sentry-hdfs/sentry-hdfs-service/.gitignore
new file mode 100644
index 0000000..91ad75b
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-service/.gitignore
@@ -0,0 +1,18 @@
+*.class
+target/
+.classpath
+.project
+.settings
+.metadata
+.idea/
+*.iml
+derby.log
+datanucleus.log
+sentry-core/sentry-core-common/src/gen
+**/TempStatsStore/
+# Package Files #
+*.jar
+*.war
+*.ear
+test-output/
+maven-repo/
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-service/pom.xml
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/pom.xml b/sentry-hdfs/sentry-hdfs-service/pom.xml
new file mode 100644
index 0000000..365380e
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-service/pom.xml
@@ -0,0 +1,104 @@
+<?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.
+-->
+<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.sentry</groupId>
+ <artifactId>sentry-hdfs</artifactId>
+ <version>1.5.0-incubating-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>sentry-hdfs-service</artifactId>
+ <name>Sentry HDFS service</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-common</artifactId>
+ <scope>provided</scope>
+ </dependency>
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>log4j</groupId>
+ <artifactId>log4j</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.shiro</groupId>
+ <artifactId>shiro-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-log4j12</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.sentry</groupId>
+ <artifactId>sentry-hdfs-common</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.sentry</groupId>
+ <artifactId>sentry-provider-db</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-exec</artifactId>
+ <version>${hive.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-shims</artifactId>
+ <version>${hive.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.thrift</groupId>
+ <artifactId>libfb303</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.thrift</groupId>
+ <artifactId>libthrift</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>ant-contrib</groupId>
+ <artifactId>ant-contrib</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hadoop</groupId>
+ <artifactId>hadoop-minikdc</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-metastore</artifactId>
+ <version>${hive.version}</version>
+ </dependency>
+ </dependencies>
+
+
+</project>
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/ExtendedMetastoreClient.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/ExtendedMetastoreClient.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/ExtendedMetastoreClient.java
new file mode 100644
index 0000000..e7677f2
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/ExtendedMetastoreClient.java
@@ -0,0 +1,108 @@
+/*
+ * 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.hdfs;
+
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Implementation of {@link MetastoreClient}
+ *
+ */
+public class ExtendedMetastoreClient implements MetastoreClient {
+
+ private static Logger LOG = LoggerFactory.getLogger(ExtendedMetastoreClient.class);
+
+ private volatile HiveMetaStoreClient client;
+ private final HiveConf hiveConf;
+ public ExtendedMetastoreClient(HiveConf hiveConf) {
+ this.hiveConf = hiveConf;
+ }
+
+ @Override
+ public List<Database> getAllDatabases() {
+ List<Database> retList = new ArrayList<Database>();
+ HiveMetaStoreClient client = getClient();
+ if (client != null) {
+ try {
+ for (String dbName : client.getAllDatabases()) {
+ retList.add(client.getDatabase(dbName));
+ }
+ } catch (Exception e) {
+ LOG.error("Could not get All Databases !!", e);
+ }
+ }
+ return retList;
+ }
+
+ @Override
+ public List<Table> getAllTablesOfDatabase(Database db) {
+ List<Table> retList = new ArrayList<Table>();
+ HiveMetaStoreClient client = getClient();
+ if (client != null) {
+ try {
+ for (String tblName : client.getAllTables(db.getName())) {
+ retList.add(client.getTable(db.getName(), tblName));
+ }
+ } catch (Exception e) {
+ LOG.error(String.format(
+ "Could not get Tables for '%s' !!", db.getName()), e);
+ }
+ }
+ return retList;
+ }
+
+ @Override
+ public List<Partition> listAllPartitions(Database db, Table tbl) {
+ HiveMetaStoreClient client = getClient();
+ if (client != null) {
+ try {
+ return client.listPartitions(db.getName(), tbl.getTableName(), Short.MAX_VALUE);
+ } catch (Exception e) {
+ LOG.error(String.format(
+ "Could not get partitions for '%s'.'%s' !!", db.getName(),
+ tbl.getTableName()), e);
+ }
+ }
+ return new LinkedList<Partition>();
+ }
+
+ private HiveMetaStoreClient getClient() {
+ if (client == null) {
+ try {
+ client = new HiveMetaStoreClient(hiveConf);
+ return client;
+ } catch (MetaException e) {
+ client = null;
+ LOG.error("Could not create metastore client !!", e);
+ return null;
+ }
+ } else {
+ return client;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/MetastorePlugin.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/MetastorePlugin.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/MetastorePlugin.java
new file mode 100644
index 0000000..9a81e3a
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/MetastorePlugin.java
@@ -0,0 +1,257 @@
+/**
+ * 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.hdfs;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.HiveMetaStore;
+import org.apache.hadoop.hive.metastore.HiveMetaStore.HMSHandler;
+import org.apache.hadoop.hive.metastore.IHMSHandler;
+import org.apache.hadoop.hive.metastore.MetaStorePreEventListener;
+import org.apache.hadoop.hive.metastore.api.Database;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.sentry.hdfs.ServiceConstants.ServerConfig;
+import org.apache.sentry.hdfs.service.thrift.TPathChanges;
+import org.apache.sentry.provider.db.SentryMetastoreListenerPlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+/**
+ * Plugin implementation of {@link SentryMetastoreListenerPlugin} that hooks
+ * into the sites in the {@link MetaStorePreEventListener} that deal with
+ * creation/updation and deletion for paths.
+ */
+public class MetastorePlugin extends SentryMetastoreListenerPlugin {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(MetastorePlugin.class);
+
+ private final Configuration conf;
+ private SentryHDFSServiceClient sentryClient;
+ private UpdateableAuthzPaths authzPaths;
+ private Lock notificiationLock;
+
+ //Initialized to some value > 1 so that the first update notification
+ // will trigger a full Image fetch
+ private final AtomicLong seqNum = new AtomicLong(5);
+ private volatile long lastSentSeqNum = -1;
+ private final ExecutorService threadPool;
+
+ static class ProxyHMSHandler extends HMSHandler {
+ public ProxyHMSHandler(String name, HiveConf conf) throws MetaException {
+ super(name, conf);
+ }
+ @Override
+ public String startFunction(String function, String extraLogInfo) {
+ return function;
+ }
+ }
+
+ public MetastorePlugin(Configuration conf) {
+ this.notificiationLock = new ReentrantLock();
+ this.conf = new HiveConf((HiveConf)conf);
+ this.conf.unset(HiveConf.ConfVars.METASTORE_PRE_EVENT_LISTENERS.varname);
+ this.conf.unset(HiveConf.ConfVars.METASTORE_EVENT_LISTENERS.varname);
+ this.conf.unset(HiveConf.ConfVars.METASTORE_END_FUNCTION_LISTENERS.varname);
+ this.conf.unset(HiveConf.ConfVars.METASTOREURIS.varname);
+ try {
+ this.authzPaths = createInitialUpdate(new ProxyHMSHandler("sentry.hdfs", (HiveConf)this.conf));
+ } catch (Exception e1) {
+ LOGGER.error("Could not create Initial AuthzPaths or HMSHandler !!", e1);
+ throw new RuntimeException(e1);
+ }
+ try {
+ sentryClient = new SentryHDFSServiceClient(conf);
+ } catch (Exception e) {
+ sentryClient = null;
+ LOGGER.error("Could not connect to Sentry HDFS Service !!", e);
+ }
+ ScheduledExecutorService threadPool = Executors.newScheduledThreadPool(1);
+ threadPool.scheduleWithFixedDelay(new Runnable() {
+ @Override
+ public void run() {
+ notificiationLock.lock();
+ try {
+ long lastSeenHMSPathSeqNum =
+ MetastorePlugin.this.getClient().getLastSeenHMSPathSeqNum();
+ if (lastSeenHMSPathSeqNum != lastSentSeqNum) {
+ LOGGER.warn("Sentry not in sync with HMS [" + lastSeenHMSPathSeqNum + ", " + lastSentSeqNum + "]");
+ PathsUpdate fullImageUpdate =
+ MetastorePlugin.this.authzPaths.createFullImageUpdate(
+ lastSentSeqNum);
+ LOGGER.warn("Sentry not in sync with HMS !!");
+ notifySentryNoLock(fullImageUpdate, false);
+ }
+ } catch (Exception e) {
+ sentryClient = null;
+ LOGGER.error("Error talking to Sentry HDFS Service !!", e);
+ } finally {
+ notificiationLock.unlock();
+ }
+ }
+ }, this.conf.getLong(ServerConfig.SENTRY_HDFS_INIT_UPDATE_RETRY_DELAY_MS,
+ ServerConfig.SENTRY_HDFS_INIT_UPDATE_RETRY_DELAY_DEFAULT), 1000,
+ TimeUnit.MILLISECONDS);
+ this.threadPool = threadPool;
+ }
+
+ private UpdateableAuthzPaths createInitialUpdate(IHMSHandler hmsHandler) throws Exception {
+ UpdateableAuthzPaths authzPaths = new UpdateableAuthzPaths(new String[] {"/"});
+ PathsUpdate tempUpdate = new PathsUpdate(-1, false);
+ List<String> allDbStr = hmsHandler.get_all_databases();
+ for (String dbName : allDbStr) {
+ Database db = hmsHandler.get_database(dbName);
+ tempUpdate.newPathChange(db.getName()).addToAddPaths(
+ PathsUpdate.cleanPath(db.getLocationUri()));
+ List<String> allTblStr = hmsHandler.get_all_tables(db.getName());
+ for (String tblName : allTblStr) {
+ Table tbl = hmsHandler.get_table(db.getName(), tblName);
+ TPathChanges tblPathChange = tempUpdate.newPathChange(tbl
+ .getDbName() + "." + tbl.getTableName());
+ List<Partition> tblParts =
+ hmsHandler.get_partitions(db.getName(), tbl.getTableName(), (short) -1);
+ tblPathChange.addToAddPaths(PathsUpdate.cleanPath(tbl.getSd()
+ .getLocation() == null ? db.getLocationUri() : tbl
+ .getSd().getLocation()));
+ for (Partition part : tblParts) {
+ tblPathChange.addToAddPaths(PathsUpdate.cleanPath(part.getSd()
+ .getLocation()));
+ }
+ }
+ }
+ authzPaths.updatePartial(Lists.newArrayList(tempUpdate),
+ new ReentrantReadWriteLock());
+ return authzPaths;
+ }
+
+ @Override
+ public void addPath(String authzObj, String path) {
+ LOGGER.debug("#### HMS Path Update ["
+ + "OP : addPath, "
+ + "authzObj : " + authzObj + ", "
+ + "path : " + path + "]");
+ PathsUpdate update = createHMSUpdate();
+ update.newPathChange(authzObj).addToAddPaths(PathsUpdate.cleanPath(path));
+ notifySentry(update, true);
+ }
+
+ @Override
+ public void removeAllPaths(String authzObj, List<String> childObjects) {
+ LOGGER.debug("#### HMS Path Update ["
+ + "OP : removeAllPaths, "
+ + "authzObj : " + authzObj + ", "
+ + "childObjs : " + (childObjects == null ? "[]" : childObjects) + "]");
+ PathsUpdate update = createHMSUpdate();
+ if (childObjects != null) {
+ for (String childObj : childObjects) {
+ update.newPathChange(authzObj + "." + childObj).addToDelPaths(
+ Lists.newArrayList(PathsUpdate.ALL_PATHS));
+ }
+ }
+ update.newPathChange(authzObj).addToDelPaths(
+ Lists.newArrayList(PathsUpdate.ALL_PATHS));
+ notifySentry(update, true);
+ }
+
+ @Override
+ public void removePath(String authzObj, String path) {
+ if ("*".equals(path)) {
+ removeAllPaths(authzObj, null);
+ } else {
+ LOGGER.debug("#### HMS Path Update ["
+ + "OP : removePath, "
+ + "authzObj : " + authzObj + ", "
+ + "path : " + path + "]");
+ PathsUpdate update = createHMSUpdate();
+ update.newPathChange(authzObj).addToDelPaths(PathsUpdate.cleanPath(path));
+ notifySentry(update, true);
+ }
+ }
+
+ @Override
+ public void renameAuthzObject(String oldName, String oldPath, String newName,
+ String newPath) {
+ PathsUpdate update = createHMSUpdate();
+ LOGGER.debug("#### HMS Path Update ["
+ + "OP : renameAuthzObject, "
+ + "oldName : " + oldName + ","
+ + "newPath : " + oldPath + ","
+ + "newName : " + newName + ","
+ + "newPath : " + newPath + "]");
+ update.newPathChange(newName).addToAddPaths(PathsUpdate.cleanPath(newPath));
+ update.newPathChange(oldName).addToDelPaths(PathsUpdate.cleanPath(oldPath));
+ notifySentry(update, true);
+ }
+
+ private SentryHDFSServiceClient getClient() {
+ if (sentryClient == null) {
+ try {
+ sentryClient = new SentryHDFSServiceClient(conf);
+ } catch (IOException e) {
+ sentryClient = null;
+ LOGGER.error("Could not connect to Sentry HDFS Service !!", e);
+ }
+ }
+ return sentryClient;
+ }
+
+ private PathsUpdate createHMSUpdate() {
+ PathsUpdate update = new PathsUpdate(seqNum.incrementAndGet(), false);
+ LOGGER.debug("#### HMS Path Update SeqNum : [" + seqNum.get() + "]");
+ return update;
+ }
+
+ private void notifySentryNoLock(PathsUpdate update, boolean applyLocal) {
+ if (applyLocal) {
+ authzPaths.updatePartial(Lists.newArrayList(update), new ReentrantReadWriteLock());
+ }
+ try {
+ getClient().notifyHMSUpdate(update);
+ } catch (Exception e) {
+ LOGGER.error("Could not send update to Sentry HDFS Service !!", e);
+ } finally {
+ lastSentSeqNum = update.getSeqNum();
+ LOGGER.debug("#### HMS Path Last update sent : [" + lastSentSeqNum + "]");
+ }
+ }
+
+ private void notifySentry(PathsUpdate update, boolean applyLocal) {
+ notificiationLock.lock();
+ try {
+ notifySentryNoLock(update, applyLocal);
+ } finally {
+ notificiationLock.unlock();
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessor.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessor.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessor.java
new file mode 100644
index 0000000..cc849b9
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessor.java
@@ -0,0 +1,118 @@
+/**
+ * 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.hdfs;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.sentry.hdfs.service.thrift.SentryHDFSService;
+import org.apache.sentry.hdfs.service.thrift.TAuthzUpdateResponse;
+import org.apache.sentry.hdfs.service.thrift.TPathsUpdate;
+import org.apache.sentry.hdfs.service.thrift.TPermissionsUpdate;
+import org.apache.thrift.TException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SentryHDFSServiceProcessor implements SentryHDFSService.Iface {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SentryHDFSServiceProcessor.class);
+
+ @Override
+ public TAuthzUpdateResponse get_all_authz_updates_from(long permSeqNum, long pathSeqNum)
+ throws TException {
+ TAuthzUpdateResponse retVal = new TAuthzUpdateResponse();
+ retVal.setAuthzPathUpdate(new LinkedList<TPathsUpdate>());
+ retVal.setAuthzPermUpdate(new LinkedList<TPermissionsUpdate>());
+ if (SentryPlugin.instance != null) {
+ List<PermissionsUpdate> permUpdates = SentryPlugin.instance.getAllPermsUpdatesFrom(permSeqNum);
+ List<PathsUpdate> pathUpdates = SentryPlugin.instance.getAllPathsUpdatesFrom(pathSeqNum);
+ try {
+ for (PathsUpdate update : pathUpdates) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("### Sending PATH preUpdate seq [" + update.getSeqNum() + "] ###");
+ LOGGER.debug("### Sending PATH preUpdate [" + update.toThrift() + "] ###");
+ }
+ retVal.getAuthzPathUpdate().add(update.toThrift());
+ }
+ for (PermissionsUpdate update : permUpdates) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("### Sending PERM preUpdate seq [" + update.getSeqNum() + "] ###");
+ LOGGER.debug("### Sending PERM preUpdate [" + update.toThrift() + "] ###");
+ }
+ retVal.getAuthzPermUpdate().add(update.toThrift());
+ }
+ if (LOGGER.isDebugEnabled()) {
+ StringBuilder permSeq = new StringBuilder("<");
+ for (PermissionsUpdate permUpdate : permUpdates) {
+ permSeq.append(permUpdate.getSeqNum()).append(",");
+ }
+ permSeq.append(">");
+ StringBuilder pathSeq = new StringBuilder("<");
+ for (PathsUpdate pathUpdate : pathUpdates) {
+ pathSeq.append(pathUpdate.getSeqNum()).append(",");
+ }
+ pathSeq.append(">");
+ LOGGER.debug("#### Updates requested from HDFS ["
+ + "permReq=" + permSeqNum + ", permResp=" + permSeq + "] "
+ + "[pathReq=" + pathSeqNum + ", pathResp=" + pathSeq + "]");
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error Sending updates to downstream Cache", e);
+ throw new TException(e);
+ }
+ } else {
+ LOGGER.error("SentryPlugin not initialized yet !!");
+ }
+
+ return retVal;
+ }
+
+ @Override
+ public void handle_hms_notification(TPathsUpdate update) throws TException {
+ try {
+ PathsUpdate hmsUpdate = new PathsUpdate(update);
+ if (SentryPlugin.instance != null) {
+ SentryPlugin.instance.handlePathUpdateNotification(hmsUpdate);
+ LOGGER.debug("Authz Paths update [" + hmsUpdate.getSeqNum() + "]..");
+ } else {
+ LOGGER.error("SentryPlugin not initialized yet !!");
+ }
+ } catch (Exception e) {
+ LOGGER.error("Error handling notification from HMS", e);
+ throw new TException(e);
+ }
+ }
+
+ @Override
+ public long check_hms_seq_num(long pathSeqNum) throws TException {
+ return SentryPlugin.instance.getLastSeenHMSPathSeqNum();
+ }
+
+ /**
+ * Not implemented for the time being..
+ */
+ @Override
+ public Map<String, List<String>> get_all_related_paths(String arg0,
+ boolean arg1) throws TException {
+ // TODO Auto-generated method stub
+ return null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessorFactory.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessorFactory.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessorFactory.java
new file mode 100644
index 0000000..d35de75
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryHDFSServiceProcessorFactory.java
@@ -0,0 +1,108 @@
+/**
+ * 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.hdfs;
+
+import java.net.Socket;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.sentry.hdfs.service.thrift.SentryHDFSService;
+import org.apache.sentry.hdfs.service.thrift.SentryHDFSService.Iface;
+import org.apache.sentry.provider.db.log.util.CommandUtil;
+import org.apache.sentry.service.thrift.ProcessorFactory;
+import org.apache.thrift.TException;
+import org.apache.thrift.TMultiplexedProcessor;
+import org.apache.thrift.TProcessor;
+import org.apache.thrift.protocol.TProtocol;
+import org.apache.thrift.transport.TSaslClientTransport;
+import org.apache.thrift.transport.TSaslServerTransport;
+import org.apache.thrift.transport.TSocket;
+import org.apache.thrift.transport.TTransport;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SentryHDFSServiceProcessorFactory extends ProcessorFactory{
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SentryHDFSServiceProcessorFactory.class);
+
+ static class ProcessorWrapper extends SentryHDFSService.Processor<SentryHDFSService.Iface> {
+
+ public ProcessorWrapper(Iface iface) {
+ super(iface);
+ }
+ @Override
+ public boolean process(TProtocol in, TProtocol out) throws TException {
+ setIpAddress(in);
+ setImpersonator(in);
+ return super.process(in, out);
+ }
+
+ private void setImpersonator(final TProtocol in) {
+ TTransport transport = in.getTransport();
+ if (transport instanceof TSaslServerTransport) {
+ String impersonator = ((TSaslServerTransport) transport).getSaslServer().getAuthorizationID();
+ CommandUtil.setImpersonator(impersonator);
+ }
+ }
+
+ private void setIpAddress(final TProtocol in) {
+ TTransport transport = in.getTransport();
+ TSocket tSocket = getUnderlyingSocketFromTransport(transport);
+ if (tSocket != null) {
+ setIpAddress(tSocket.getSocket());
+ } else {
+ LOGGER.warn("Unknown Transport, cannot determine ipAddress");
+ }
+ }
+
+ private void setIpAddress(Socket socket) {
+ CommandUtil.setIpAddress(socket.getInetAddress().toString());
+ }
+
+ private TSocket getUnderlyingSocketFromTransport(TTransport transport) {
+ if (transport != null) {
+ if (transport instanceof TSaslServerTransport) {
+ transport = ((TSaslServerTransport) transport).getUnderlyingTransport();
+ } else if (transport instanceof TSaslClientTransport) {
+ transport = ((TSaslClientTransport) transport).getUnderlyingTransport();
+ } else {
+ if (!(transport instanceof TSocket)) {
+ LOGGER.warn("Transport class [" + transport.getClass().getName() + "] is not of type TSocket");
+ return null;
+ }
+ }
+ return (TSocket) transport;
+ }
+ return null;
+ }
+ }
+
+ public SentryHDFSServiceProcessorFactory(Configuration conf) {
+ super(conf);
+ }
+
+
+ public boolean register(TMultiplexedProcessor multiplexedProcessor) throws Exception {
+ SentryHDFSServiceProcessor sentryServiceHandler =
+ new SentryHDFSServiceProcessor();
+ TProcessor processor = new ProcessorWrapper(sentryServiceHandler);
+ multiplexedProcessor.registerProcessor(
+ SentryHDFSServiceClient.SENTRY_HDFS_SERVICE_NAME, processor);
+ return true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryPlugin.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryPlugin.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryPlugin.java
new file mode 100644
index 0000000..55b7697
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/SentryPlugin.java
@@ -0,0 +1,247 @@
+/**
+ * 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.hdfs;
+
+import java.util.HashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.sentry.hdfs.ServiceConstants.ServerConfig;
+import org.apache.sentry.hdfs.UpdateForwarder.ExternalImageRetriever;
+import org.apache.sentry.hdfs.service.thrift.TPathChanges;
+import org.apache.sentry.hdfs.service.thrift.TPermissionsUpdate;
+import org.apache.sentry.hdfs.service.thrift.TPrivilegeChanges;
+import org.apache.sentry.hdfs.service.thrift.TRoleChanges;
+import org.apache.sentry.provider.db.SentryPolicyStorePlugin;
+import org.apache.sentry.provider.db.SentryPolicyStorePlugin.SentryPluginException;
+import org.apache.sentry.provider.db.service.persistent.SentryStore;
+import org.apache.sentry.provider.db.service.thrift.TAlterSentryRoleAddGroupsRequest;
+import org.apache.sentry.provider.db.service.thrift.TAlterSentryRoleDeleteGroupsRequest;
+import org.apache.sentry.provider.db.service.thrift.TAlterSentryRoleGrantPrivilegeRequest;
+import org.apache.sentry.provider.db.service.thrift.TAlterSentryRoleRevokePrivilegeRequest;
+import org.apache.sentry.provider.db.service.thrift.TDropPrivilegesRequest;
+import org.apache.sentry.provider.db.service.thrift.TDropSentryRoleRequest;
+import org.apache.sentry.provider.db.service.thrift.TRenamePrivilegesRequest;
+import org.apache.sentry.provider.db.service.thrift.TSentryAuthorizable;
+import org.apache.sentry.provider.db.service.thrift.TSentryGroup;
+import org.apache.sentry.provider.db.service.thrift.TSentryPrivilege;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Strings;
+
+public class SentryPlugin implements SentryPolicyStorePlugin {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SentryPlugin.class);
+
+ public static volatile SentryPlugin instance;
+
+ static class PermImageRetriever implements ExternalImageRetriever<PermissionsUpdate> {
+
+ private final SentryStore sentryStore;
+
+ public PermImageRetriever(SentryStore sentryStore) {
+ this.sentryStore = sentryStore;
+ }
+
+ @Override
+ public PermissionsUpdate retrieveFullImage(long currSeqNum) {
+ Map<String, HashMap<String, String>> privilegeImage = sentryStore.retrieveFullPrivilegeImage();
+ Map<String, LinkedList<String>> roleImage = sentryStore.retrieveFullRoleImage();
+
+ TPermissionsUpdate tPermUpdate = new TPermissionsUpdate(true, currSeqNum,
+ new HashMap<String, TPrivilegeChanges>(),
+ new HashMap<String, TRoleChanges>());
+ for (Map.Entry<String, HashMap<String, String>> privEnt : privilegeImage.entrySet()) {
+ String authzObj = privEnt.getKey();
+ HashMap<String,String> privs = privEnt.getValue();
+ tPermUpdate.putToPrivilegeChanges(authzObj, new TPrivilegeChanges(
+ authzObj, privs, new HashMap<String, String>()));
+ }
+ for (Map.Entry<String, LinkedList<String>> privEnt : roleImage.entrySet()) {
+ String role = privEnt.getKey();
+ LinkedList<String> groups = privEnt.getValue();
+ tPermUpdate.putToRoleChanges(role, new TRoleChanges(role, groups, new LinkedList<String>()));
+ }
+ PermissionsUpdate permissionsUpdate = new PermissionsUpdate(tPermUpdate);
+ permissionsUpdate.setSeqNum(currSeqNum);
+ return permissionsUpdate;
+ }
+
+ }
+
+ private UpdateForwarder<PathsUpdate> pathsUpdater;
+ private UpdateForwarder<PermissionsUpdate> permsUpdater;
+ private final AtomicLong permSeqNum = new AtomicLong(5);
+ private PermImageRetriever permImageRetriever;
+
+ long getLastSeenHMSPathSeqNum() {
+ return pathsUpdater.getLastSeen();
+ }
+
+ @Override
+ public void initialize(Configuration conf, SentryStore sentryStore) throws SentryPluginException {
+ final String[] pathPrefixes = conf
+ .getStrings(ServerConfig.SENTRY_HDFS_INTEGRATION_PATH_PREFIXES,
+ ServerConfig.SENTRY_HDFS_INTEGRATION_PATH_PREFIXES_DEFAULT);
+ final int initUpdateRetryDelayMs =
+ conf.getInt(ServerConfig.SENTRY_HDFS_INIT_UPDATE_RETRY_DELAY_MS,
+ ServerConfig.SENTRY_HDFS_INIT_UPDATE_RETRY_DELAY_DEFAULT);
+ pathsUpdater = new UpdateForwarder<PathsUpdate>(new UpdateableAuthzPaths(
+ pathPrefixes), null, 100, initUpdateRetryDelayMs);
+ permImageRetriever = new PermImageRetriever(sentryStore);
+ permsUpdater = new UpdateForwarder<PermissionsUpdate>(
+ new UpdateablePermissions(permImageRetriever), permImageRetriever,
+ 100, initUpdateRetryDelayMs);
+ LOGGER.info("Sentry HDFS plugin initialized !!");
+ instance = this;
+ }
+
+ public List<PathsUpdate> getAllPathsUpdatesFrom(long pathSeqNum) {
+ return pathsUpdater.getAllUpdatesFrom(pathSeqNum);
+ }
+
+ public List<PermissionsUpdate> getAllPermsUpdatesFrom(long permSeqNum) {
+ return permsUpdater.getAllUpdatesFrom(permSeqNum);
+ }
+
+ public void handlePathUpdateNotification(PathsUpdate update) {
+ pathsUpdater.handleUpdateNotification(update);
+ LOGGER.debug("Recieved Authz Path update [" + update.getSeqNum() + "]..");
+ }
+
+ @Override
+ public void onAlterSentryRoleAddGroups(
+ TAlterSentryRoleAddGroupsRequest request) throws SentryPluginException {
+ PermissionsUpdate update = new PermissionsUpdate(permSeqNum.incrementAndGet(), false);
+ TRoleChanges rUpdate = update.addRoleUpdate(request.getRoleName());
+ for (TSentryGroup group : request.getGroups()) {
+ rUpdate.addToAddGroups(group.getGroupName());
+ }
+ permsUpdater.handleUpdateNotification(update);
+ LOGGER.debug("Authz Perm preUpdate [" + update.getSeqNum() + ", " + request.getRoleName() + "]..");
+ }
+
+ @Override
+ public void onAlterSentryRoleDeleteGroups(
+ TAlterSentryRoleDeleteGroupsRequest request)
+ throws SentryPluginException {
+ PermissionsUpdate update = new PermissionsUpdate(permSeqNum.incrementAndGet(), false);
+ TRoleChanges rUpdate = update.addRoleUpdate(request.getRoleName());
+ for (TSentryGroup group : request.getGroups()) {
+ rUpdate.addToDelGroups(group.getGroupName());
+ }
+ permsUpdater.handleUpdateNotification(update);
+ LOGGER.debug("Authz Perm preUpdate [" + update.getSeqNum() + ", " + request.getRoleName() + "]..");
+ }
+
+ @Override
+ public void onAlterSentryRoleGrantPrivilege(
+ TAlterSentryRoleGrantPrivilegeRequest request)
+ throws SentryPluginException {
+ String authzObj = getAuthzObj(request.getPrivilege());
+ if (authzObj != null) {
+ PermissionsUpdate update = new PermissionsUpdate(permSeqNum.incrementAndGet(), false);
+ update.addPrivilegeUpdate(authzObj).putToAddPrivileges(
+ request.getRoleName(), request.getPrivilege().getAction().toUpperCase());
+ permsUpdater.handleUpdateNotification(update);
+ LOGGER.debug("Authz Perm preUpdate [" + update.getSeqNum() + "]..");
+ }
+ }
+
+ @Override
+ public void onRenameSentryPrivilege(TRenamePrivilegesRequest request)
+ throws SentryPluginException {
+ String oldAuthz = getAuthzObj(request.getOldAuthorizable());
+ String newAuthz = getAuthzObj(request.getNewAuthorizable());
+ PermissionsUpdate update = new PermissionsUpdate(permSeqNum.incrementAndGet(), false);
+ TPrivilegeChanges privUpdate = update.addPrivilegeUpdate(PermissionsUpdate.RENAME_PRIVS);
+ privUpdate.putToAddPrivileges(newAuthz, newAuthz);
+ privUpdate.putToDelPrivileges(oldAuthz, oldAuthz);
+ permsUpdater.handleUpdateNotification(update);
+ LOGGER.debug("Authz Perm preUpdate [" + update.getSeqNum() + ", " + newAuthz + ", " + oldAuthz + "]..");
+ }
+
+ @Override
+ public void onAlterSentryRoleRevokePrivilege(
+ TAlterSentryRoleRevokePrivilegeRequest request)
+ throws SentryPluginException {
+ String authzObj = getAuthzObj(request.getPrivilege());
+ if (authzObj != null) {
+ PermissionsUpdate update = new PermissionsUpdate(permSeqNum.incrementAndGet(), false);
+ update.addPrivilegeUpdate(authzObj).putToDelPrivileges(
+ request.getRoleName(), request.getPrivilege().getAction().toUpperCase());
+ permsUpdater.handleUpdateNotification(update);
+ LOGGER.debug("Authz Perm preUpdate [" + update.getSeqNum() + ", " + authzObj + "]..");
+ }
+ }
+
+ @Override
+ public void onDropSentryRole(TDropSentryRoleRequest request)
+ throws SentryPluginException {
+ PermissionsUpdate update = new PermissionsUpdate(permSeqNum.incrementAndGet(), false);
+ update.addPrivilegeUpdate(PermissionsUpdate.ALL_AUTHZ_OBJ).putToDelPrivileges(
+ request.getRoleName(), PermissionsUpdate.ALL_AUTHZ_OBJ);
+ update.addRoleUpdate(request.getRoleName()).addToDelGroups(PermissionsUpdate.ALL_GROUPS);
+ permsUpdater.handleUpdateNotification(update);
+ LOGGER.debug("Authz Perm preUpdate [" + update.getSeqNum() + ", " + request.getRoleName() + "]..");
+ }
+
+ @Override
+ public void onDropSentryPrivilege(TDropPrivilegesRequest request)
+ throws SentryPluginException {
+ PermissionsUpdate update = new PermissionsUpdate(permSeqNum.incrementAndGet(), false);
+ String authzObj = getAuthzObj(request.getAuthorizable());
+ update.addPrivilegeUpdate(authzObj).putToDelPrivileges(
+ PermissionsUpdate.ALL_ROLES, PermissionsUpdate.ALL_ROLES);
+ permsUpdater.handleUpdateNotification(update);
+ LOGGER.debug("Authz Perm preUpdate [" + update.getSeqNum() + ", " + authzObj + "]..");
+ }
+
+ private String getAuthzObj(TSentryPrivilege privilege) {
+ String authzObj = null;
+ if (!SentryStore.isNULL(privilege.getDbName())) {
+ String dbName = privilege.getDbName();
+ String tblName = privilege.getTableName();
+ if (SentryStore.isNULL(tblName)) {
+ authzObj = dbName;
+ } else {
+ authzObj = dbName + "." + tblName;
+ }
+ }
+ return authzObj;
+ }
+
+ private String getAuthzObj(TSentryAuthorizable authzble) {
+ String authzObj = null;
+ if (!SentryStore.isNULL(authzble.getDb())) {
+ String dbName = authzble.getDb();
+ String tblName = authzble.getTable();
+ if (SentryStore.isNULL(tblName)) {
+ authzObj = dbName;
+ } else {
+ authzObj = dbName + "." + tblName;
+ }
+ }
+ return authzObj;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-sentry/blob/2e509e4b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/UpdateForwarder.java
----------------------------------------------------------------------
diff --git a/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/UpdateForwarder.java b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/UpdateForwarder.java
new file mode 100644
index 0000000..f321d3d
--- /dev/null
+++ b/sentry-hdfs/sentry-hdfs-service/src/main/java/org/apache/sentry/hdfs/UpdateForwarder.java
@@ -0,0 +1,292 @@
+/**
+ * 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.hdfs;
+
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.collect.Lists;
+
+public class UpdateForwarder<K extends Updateable.Update> implements
+ Updateable<K> {
+
+ public static interface ExternalImageRetriever<K> {
+
+ public K retrieveFullImage(long currSeqNum);
+
+ }
+
+ private final AtomicLong lastSeenSeqNum = new AtomicLong(0);
+ private final AtomicLong lastCommittedSeqNum = new AtomicLong(0);
+ // Updates should be handled in order
+ private final Executor updateHandler = Executors.newSingleThreadExecutor();
+
+ // Update log is used when propagate updates to a downstream cache.
+ // The preUpdate log stores all commits that were applied to this cache.
+ // When the update log is filled to capacity (updateLogSize), all
+ // entries are cleared and a compact image if the state of the cache is
+ // appended to the log.
+ // The first entry in an update log (consequently the first preUpdate a
+ // downstream cache sees) will be a full image. All subsequent entries are
+ // partial edits
+ private final LinkedList<K> updateLog = new LinkedList<K>();
+ // UpdateLog is disabled when updateLogSize = 0;
+ private final int updateLogSize;
+
+ private final ExternalImageRetriever<K> imageRetreiver;
+
+ private volatile Updateable<K> updateable;
+
+ private final ReadWriteLock lock = new ReentrantReadWriteLock();
+ private static final long INIT_SEQ_NUM = -2;
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(UpdateForwarder.class);
+
+ public UpdateForwarder(Updateable<K> updateable,
+ ExternalImageRetriever<K> imageRetreiver, int updateLogSize) {
+ this(updateable, imageRetreiver, updateLogSize, 5000);
+ }
+ public UpdateForwarder(Updateable<K> updateable,
+ ExternalImageRetriever<K> imageRetreiver, int updateLogSize,
+ int initUpdateRetryDelay) {
+ this.updateLogSize = updateLogSize;
+ this.imageRetreiver = imageRetreiver;
+ if (imageRetreiver != null) {
+ spawnInitialUpdater(updateable, initUpdateRetryDelay);
+ } else {
+ this.updateable = updateable;
+ }
+ }
+
+ private void spawnInitialUpdater(final Updateable<K> updateable,
+ final int initUpdateRetryDelay) {
+ K firstFullImage = null;
+ try {
+ firstFullImage = imageRetreiver.retrieveFullImage(INIT_SEQ_NUM);
+ } catch (Exception e) {
+ LOGGER.warn("InitialUpdater encountered exception !! ", e);
+ firstFullImage = null;
+ Thread initUpdater = new Thread() {
+ @Override
+ public void run() {
+ while (UpdateForwarder.this.updateable == null) {
+ try {
+ Thread.sleep(initUpdateRetryDelay);
+ } catch (InterruptedException e) {
+ LOGGER.warn("Thread interrupted !! ", e);
+ break;
+ }
+ K fullImage = null;
+ try {
+ fullImage =
+ UpdateForwarder.this.imageRetreiver
+ .retrieveFullImage(INIT_SEQ_NUM);
+ appendToUpdateLog(fullImage);
+ } catch (Exception e) {
+ LOGGER.warn("InitialUpdater encountered exception !! ", e);
+ }
+ if (fullImage != null) {
+ UpdateForwarder.this.updateable = updateable.updateFull(fullImage);
+ }
+ }
+ }
+ };
+ initUpdater.start();
+ }
+ if (firstFullImage != null) {
+ appendToUpdateLog(firstFullImage);
+ this.updateable = updateable.updateFull(firstFullImage);
+ }
+ }
+ /**
+ * Handle notifications from HMS plug-in or upstream Cache
+ * @param update
+ */
+ public void handleUpdateNotification(final K update) {
+ // Correct the seqNums on the first update
+ if (lastCommittedSeqNum.get() == INIT_SEQ_NUM) {
+ K firstUpdate = updateLog.peek();
+ long firstSeqNum = update.getSeqNum() - 1;
+ if (firstUpdate != null) {
+ firstUpdate.setSeqNum(firstSeqNum);
+ }
+ lastCommittedSeqNum.set(firstSeqNum);
+ lastSeenSeqNum.set(firstSeqNum);
+ }
+ final boolean editNotMissed =
+ lastSeenSeqNum.incrementAndGet() == update.getSeqNum();
+ if (!editNotMissed) {
+ lastSeenSeqNum.set(update.getSeqNum());
+ }
+ Runnable task = new Runnable() {
+ @Override
+ public void run() {
+ K toUpdate = update;
+ if (update.hasFullImage()) {
+ updateable = updateable.updateFull(update);
+ } else {
+ if (editNotMissed) {
+ // apply partial preUpdate
+ updateable.updatePartial(Lists.newArrayList(update), lock);
+ } else {
+ // Retrieve full update from External Source and
+ if (imageRetreiver != null) {
+ toUpdate = imageRetreiver
+ .retrieveFullImage(update.getSeqNum());
+ updateable = updateable.updateFull(toUpdate);
+ }
+ }
+ }
+ appendToUpdateLog(toUpdate);
+ }
+ };
+ updateHandler.execute(task);
+ }
+
+ private void appendToUpdateLog(K update) {
+ synchronized (updateLog) {
+ boolean logCompacted = false;
+ if (updateLogSize > 0) {
+ if (update.hasFullImage() || (updateLog.size() == updateLogSize)) {
+ // Essentially a log compaction
+ updateLog.clear();
+ updateLog.add(update.hasFullImage() ? update
+ : createFullImageUpdate(update.getSeqNum()));
+ logCompacted = true;
+ } else {
+ updateLog.add(update);
+ }
+ }
+ lastCommittedSeqNum.set(update.getSeqNum());
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug("#### Appending to Update Log ["
+ + "type=" + update.getClass() + ", "
+ + "lastCommit=" + lastCommittedSeqNum.get() + ", "
+ + "lastSeen=" + lastSeenSeqNum.get() + ", "
+ + "logCompacted=" + logCompacted + "]");
+ }
+ }
+ }
+
+ /**
+ * Return all updates from requested seqNum (inclusive)
+ * @param seqNum
+ * @return
+ */
+ public List<K> getAllUpdatesFrom(long seqNum) {
+ List<K> retVal = new LinkedList<K>();
+ synchronized (updateLog) {
+ long currSeqNum = lastCommittedSeqNum.get();
+ if (LOGGER.isDebugEnabled() && (updateable != null)) {
+ LOGGER.debug("#### GetAllUpdatesFrom ["
+ + "type=" + updateable.getClass() + ", "
+ + "reqSeqNum=" + seqNum + ", "
+ + "lastCommit=" + lastCommittedSeqNum.get() + ", "
+ + "lastSeen=" + lastSeenSeqNum.get() + ", "
+ + "updateLogSize=" + updateLog.size() + "]");
+ }
+ if (updateLogSize == 0) {
+ // no updatelog configured..
+ return retVal;
+ }
+ K head = updateLog.peek();
+ if (head == null) {
+ return retVal;
+ }
+ if (seqNum > currSeqNum + 1) {
+ // This process has probably restarted since downstream
+ // recieved last update
+ retVal.addAll(updateLog);
+ return retVal;
+ }
+ if (head.getSeqNum() > seqNum) {
+ // Caller has diverged greatly..
+ if (head.hasFullImage()) {
+ // head is a refresh(full) image
+ // Send full image along with partial updates
+ for (K u : updateLog) {
+ retVal.add(u);
+ }
+ } else {
+ // Create a full image
+ // clear updateLog
+ // add fullImage to head of Log
+ // NOTE : This should ideally never happen
+ K fullImage = createFullImageUpdate(currSeqNum);
+ updateLog.clear();
+ updateLog.add(fullImage);
+ retVal.add(fullImage);
+ }
+ } else {
+ // increment iterator to requested seqNum
+ Iterator<K> iter = updateLog.iterator();
+ while (iter.hasNext()) {
+ K elem = iter.next();
+ if (elem.getSeqNum() >= seqNum) {
+ retVal.add(elem);
+ }
+ }
+ }
+ }
+ return retVal;
+ }
+
+ public boolean areAllUpdatesCommited() {
+ return lastCommittedSeqNum.get() == lastSeenSeqNum.get();
+ }
+
+ public long getLastCommitted() {
+ return lastCommittedSeqNum.get();
+ }
+
+ public long getLastSeen() {
+ return lastSeenSeqNum.get();
+ }
+
+ @Override
+ public Updateable<K> updateFull(K update) {
+ return (updateable != null) ? updateable.updateFull(update) : null;
+ }
+
+ @Override
+ public void updatePartial(Iterable<K> updates, ReadWriteLock lock) {
+ if (updateable != null) {
+ updateable.updatePartial(updates, lock);
+ }
+ }
+
+ @Override
+ public long getLastUpdatedSeqNum() {
+ return (updateable != null) ? updateable.getLastUpdatedSeqNum() : INIT_SEQ_NUM;
+ }
+
+ @Override
+ public K createFullImageUpdate(long currSeqNum) {
+ return (updateable != null) ? updateable.createFullImageUpdate(currSeqNum) : null;
+ }
+
+}