You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ranger.apache.org by ma...@apache.org on 2016/01/10 09:15:44 UTC
[1/2] incubator-ranger git commit: RANGER-807: TagSync should support
periodic full sync with Apache Atlas
Repository: incubator-ranger
Updated Branches:
refs/heads/master fec84603f -> 5b9c094ff
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasNotificationMapper.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasNotificationMapper.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasNotificationMapper.java
index 666c2c8..803a8a9 100644
--- a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasNotificationMapper.java
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasNotificationMapper.java
@@ -23,76 +23,57 @@ import org.apache.atlas.AtlasException;
import org.apache.atlas.notification.entity.EntityNotification;
import org.apache.atlas.typesystem.IReferenceableInstance;
import org.apache.atlas.typesystem.IStruct;
+import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
-import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.ranger.plugin.model.RangerPolicy;
import org.apache.ranger.plugin.model.RangerServiceResource;
import org.apache.ranger.plugin.model.RangerTag;
import org.apache.ranger.plugin.model.RangerTagDef;
import org.apache.ranger.plugin.util.ServiceTags;
-import org.apache.ranger.tagsync.process.TagSyncConfig;
import java.util.*;
public class AtlasNotificationMapper {
private static final Log LOG = LogFactory.getLog(AtlasNotificationMapper.class);
- public static final String ENTITY_TYPE_HIVE_DB = "hive_db";
- public static final String ENTITY_TYPE_HIVE_TABLE = "hive_table";
- public static final String ENTITY_TYPE_HIVE_COLUMN = "hive_column";
-
- public static final String RANGER_TYPE_HIVE_DB = "database";
- public static final String RANGER_TYPE_HIVE_TABLE = "table";
- public static final String RANGER_TYPE_HIVE_COLUMN = "column";
-
- public static final String ENTITY_ATTRIBUTE_QUALIFIED_NAME = "qualifiedName";
- public static final String ENTITY_ATTRIBUTE_QUALIFIED_NAME_FOR_HIVE_TABLE = "name";
- public static final String QUALIFIED_NAME_FORMAT_DELIMITER_STRING = "\\.";
- public static final String QUALIFIED_NAME_FORMAT_CLUSTER_DELIMITER_STRING = "@";
-
- private static Properties properties = null;
-
- public static ServiceTags processEntityNotification(EntityNotification entityNotification, Properties props) {
+ public static ServiceTags processEntityNotification(EntityNotification entityNotification) {
ServiceTags ret = null;
- properties = props;
- try {
- IReferenceableInstance entity = entityNotification.getEntity();
-
- if (isEntityMappable(entity)) {
- ret = createServiceTags(entityNotification);
- } else {
- if(LOG.isDebugEnabled()) {
- LOG.debug("Ranger not interested in Entity Notification for entity-type " + entityNotification.getEntity().getTypeName());
+ if (isNotificationHandled(entityNotification)) {
+ try {
+ IReferenceableInstance entity = entityNotification.getEntity();
+
+ if (AtlasResourceMapperUtil.isEntityTypeHandled(entity.getTypeName())) {
+ AtlasEntityWithTraits entityWithTraits = new AtlasEntityWithTraits(entityNotification.getEntity(), entityNotification.getAllTraits());
+ ret = buildServiceTags(entityWithTraits, 1L, 1L, null);
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Ranger not interested in Entity Notification for entity-type " + entityNotification.getEntity().getTypeName());
+ }
}
+ } catch (Exception exception) {
+ LOG.error("createServiceTags() failed!! ", exception);
}
- } catch (Exception exception) {
- LOG.error("createServiceTags() failed!! ", exception);
}
return ret;
}
- static private boolean isEntityMappable(IReferenceableInstance entity) {
- boolean ret = false;
-
- String entityTypeName = entity.getTypeName();
+ public static Map<String, ServiceTags> processEntitiesWithTraits(List<AtlasEntityWithTraits> atlasEntities) {
+ Map<String, ServiceTags> ret = null;
- if (StringUtils.isNotBlank(entityTypeName)) {
- if (StringUtils.equals(entityTypeName, ENTITY_TYPE_HIVE_DB) ||
- StringUtils.equals(entityTypeName, ENTITY_TYPE_HIVE_TABLE) ||
- StringUtils.equals(entityTypeName, ENTITY_TYPE_HIVE_COLUMN)) {
- ret = true;
- }
+ try {
+ ret = buildServiceTags(atlasEntities);
+ } catch (Exception exception) {
+ LOG.error("Failed to build serviceTags", exception);
}
+
return ret;
}
- static private ServiceTags createServiceTags(EntityNotification entityNotification) throws Exception {
-
- ServiceTags ret = null;
+ static private boolean isNotificationHandled(EntityNotification entityNotification) {
+ boolean ret = false;
EntityNotification.OperationType opType = entityNotification.getOperationType();
@@ -101,17 +82,10 @@ public class AtlasNotificationMapper {
LOG.debug("ENTITY_CREATE notification is not handled, as Ranger will get necessary information from any subsequent TRAIT_ADDED notification");
break;
}
- case ENTITY_UPDATE: {
- ret = getServiceTags(entityNotification);
- if (MapUtils.isEmpty(ret.getTags())) {
- LOG.debug("No traits associated with this entity update notification. Ignoring it altogether");
- ret = null;
- }
- break;
- }
+ case ENTITY_UPDATE:
case TRAIT_ADD:
case TRAIT_DELETE: {
- ret = getServiceTags(entityNotification);
+ ret = true;
break;
}
default:
@@ -121,130 +95,116 @@ public class AtlasNotificationMapper {
return ret;
}
- static private ServiceTags getServiceTags(EntityNotification entityNotification) throws Exception {
- ServiceTags ret = null;
+ static private Map<String, ServiceTags> buildServiceTags(List<AtlasEntityWithTraits> entitiesWithTraits) throws Exception {
- IReferenceableInstance entity = entityNotification.getEntity();
+ Map<String, ServiceTags> ret = new HashMap<String, ServiceTags>();
- List<RangerServiceResource> serviceResources = new ArrayList<RangerServiceResource>();
+ long serviceResourceIndex = 1L;
+ long tagIndex = 1L;
- RangerServiceResource serviceResource = getServiceResource(entity);
- serviceResources.add(serviceResource);
+ for (AtlasEntityWithTraits element : entitiesWithTraits) {
- Map<Long, RangerTag> tags = getTags(entityNotification);
+ ServiceTags serviceTags = buildServiceTags(element, serviceResourceIndex, tagIndex, ret);
- Map<Long, RangerTagDef> tagDefs = getTagDefs(tags);
+ serviceResourceIndex++;
- Map<Long, List<Long>> resourceIdToTagIds = null;
+ tagIndex += CollectionUtils.size(serviceTags.getTags());
- resourceIdToTagIds = new HashMap<Long, List<Long>>();
- List<Long> tagList = new ArrayList<Long>();
+ }
+ // Remove duplicate tag definitions
+ for (Map.Entry<String, ServiceTags> serviceTagsMapEntry : ret.entrySet()){
- if (MapUtils.isNotEmpty(tags)) {
- resourceIdToTagIds = new HashMap<Long, List<Long>>();
+ Map<Long, RangerTagDef> allTagDefs = serviceTagsMapEntry.getValue().getTagDefinitions();
- for (Map.Entry<Long, RangerTag> entry : tags.entrySet()) {
- tagList.add(entry.getKey());
+ Map<String, String> tagTypeIndex = new HashMap<String, String>();
+ Map<Long, RangerTagDef> uniqueTagDefs = new HashMap<Long, RangerTagDef>();
+
+ for (Map.Entry<Long, RangerTagDef> entry : allTagDefs.entrySet()) {
+ String tagTypeName = entry.getValue().getName();
+
+ if (tagTypeIndex.get(tagTypeName) == null) {
+ tagTypeIndex.put(tagTypeName, tagTypeName);
+ uniqueTagDefs.put(entry.getKey(), entry.getValue());
+ }
}
+ serviceTagsMapEntry.getValue().setTagDefinitions(uniqueTagDefs);
}
- resourceIdToTagIds.put(1L, tagList);
+ return ret;
+ }
+
+ static private ServiceTags buildServiceTags(AtlasEntityWithTraits entityWithTraits, long index, long tagIndex, Map<String, ServiceTags> serviceTagsMap) throws Exception {
+ ServiceTags ret = null;
- ret = new ServiceTags();
+ IReferenceableInstance entity = entityWithTraits.getEntity();
- ret.setOp(ServiceTags.OP_ADD_OR_UPDATE);
- ret.setTagModel(ServiceTags.TAGMODEL_RESOURCE_PRIVATE);
- ret.setServiceName(serviceResource.getServiceName());
- ret.setServiceResources(serviceResources);
- ret.setTagDefinitions(tagDefs);
- ret.setTags(tags);
- ret.setResourceToTagIds(resourceIdToTagIds);
+ RangerServiceResource serviceResource = AtlasResourceMapperUtil.getRangerServiceResource(entity);
- return ret;
- }
+ if (serviceResource != null) {
+ serviceResource.setId(index);
- static private RangerServiceResource getServiceResource(IReferenceableInstance entity) throws Exception {
+ String serviceName = serviceResource.getServiceName();
- RangerServiceResource ret = null;
+ Map<Long, RangerTag> tags = getTags(entityWithTraits, tagIndex);
- Map<String, RangerPolicy.RangerPolicyResource> elements = null;
- String serviceName = null;
+ Map<Long, RangerTagDef> tagDefs = getTagDefs(tags);
+ Map<Long, List<Long>> resourceIdToTagIds = null;
- elements = new HashMap<String, RangerPolicy.RangerPolicyResource>();
+ resourceIdToTagIds = new HashMap<Long, List<Long>>();
+ List<Long> tagList = new ArrayList<Long>();
+
+ if (MapUtils.isNotEmpty(tags)) {
+ resourceIdToTagIds = new HashMap<Long, List<Long>>();
- List<String> components = getQualifiedNameComponents(entity);
- // components should contain qualifiedName, clusterName, dbName, tableName, columnName in that order
+ for (Map.Entry<Long, RangerTag> entry : tags.entrySet()) {
+ tagList.add(entry.getKey());
+ }
+ }
- String entityTypeName = entity.getTypeName();
+ resourceIdToTagIds.put(index, tagList);
- String qualifiedName = components.get(0);
+ ret = createOrGetServiceTags(serviceTagsMap, serviceName);
- String clusterName, dbName, tableName, columnName;
+ ret.getServiceResources().add(serviceResource);
+ ret.getTagDefinitions().putAll(tagDefs);
+ ret.getTags().putAll(tags);
+ ret.getResourceToTagIds().putAll(resourceIdToTagIds);
- if (components.size() > 1) {
- clusterName = components.get(1);
- serviceName = getServiceName(clusterName, entityTypeName);
+ } else {
+ LOG.error("AtlasResourceMapper not found for entity-type:" + entity.getTypeName());
}
- if (StringUtils.equals(entityTypeName, ENTITY_TYPE_HIVE_DB)) {
- if (components.size() > 2) {
- dbName = components.get(2);
- RangerPolicy.RangerPolicyResource dbPolicyResource = new RangerPolicy.RangerPolicyResource(dbName);
- elements.put(RANGER_TYPE_HIVE_DB, dbPolicyResource);
+ return ret;
+ }
+
+ static private ServiceTags createOrGetServiceTags(Map<String, ServiceTags> serviceTagsMap, String serviceName) {
+ ServiceTags ret = serviceTagsMap == null ? null : serviceTagsMap.get(serviceName);
- } else {
- LOG.error("invalid qualifiedName for HIVE_DB, qualifiedName=" + qualifiedName);
- }
- } else if (StringUtils.equals(entityTypeName, ENTITY_TYPE_HIVE_TABLE)) {
- if (components.size() > 3) {
- dbName = components.get(2);
- tableName = components.get(3);
- RangerPolicy.RangerPolicyResource dbPolicyResource = new RangerPolicy.RangerPolicyResource(dbName);
- elements.put(RANGER_TYPE_HIVE_DB, dbPolicyResource);
- RangerPolicy.RangerPolicyResource tablePolicyResource = new RangerPolicy.RangerPolicyResource(tableName);
- elements.put(RANGER_TYPE_HIVE_TABLE, tablePolicyResource);
- } else {
- LOG.error("invalid qualifiedName for HIVE_TABLE, qualifiedName=" + qualifiedName);
- }
- } else if (StringUtils.equals(entityTypeName, ENTITY_TYPE_HIVE_COLUMN)) {
- if (components.size() > 4) {
- dbName = components.get(2);
- tableName = components.get(3);
- columnName = components.get(4);
- RangerPolicy.RangerPolicyResource dbPolicyResource = new RangerPolicy.RangerPolicyResource(dbName);
- elements.put(RANGER_TYPE_HIVE_DB, dbPolicyResource);
- RangerPolicy.RangerPolicyResource tablePolicyResource = new RangerPolicy.RangerPolicyResource(tableName);
- elements.put(RANGER_TYPE_HIVE_TABLE, tablePolicyResource);
- RangerPolicy.RangerPolicyResource columnPolicyResource = new RangerPolicy.RangerPolicyResource(columnName);
- elements.put(RANGER_TYPE_HIVE_COLUMN, columnPolicyResource);
- } else {
- LOG.error("invalid qualifiedName for HIVE_COLUMN, qualifiedName=" + qualifiedName);
+ if (ret == null) {
+ ret = new ServiceTags();
+
+ if (serviceTagsMap != null) {
+ serviceTagsMap.put(serviceName, ret);
}
+ ret.setOp(ServiceTags.OP_ADD_OR_UPDATE);
+ ret.setTagModel(ServiceTags.TAGMODEL_RESOURCE_PRIVATE);
+ ret.setServiceName(serviceName);
}
-
- ret = new RangerServiceResource();
- ret.setGuid(entity.getId()._getId());
- ret.setId(1L);
- ret.setServiceName(serviceName);
- ret.setResourceElements(elements);
-
return ret;
}
- static private Map<Long, RangerTag> getTags(EntityNotification entityNotification) {
+ static private Map<Long, RangerTag> getTags(AtlasEntityWithTraits entityWithTraits, long index) {
Map<Long, RangerTag> ret = null;
ret = new HashMap<Long, RangerTag>();
- long index = 1;
-
- List<IStruct> traits = entityNotification.getAllTraits();
+ List<IStruct> traits = entityWithTraits.getAllTraits();
for (IStruct trait : traits) {
@@ -288,113 +248,18 @@ public class AtlasNotificationMapper {
if (MapUtils.isNotEmpty(tags)) {
ret = new HashMap<Long, RangerTagDef>();
+
for (Map.Entry<Long, RangerTag> entry : tags.entrySet()) {
+
RangerTagDef tagDef = new RangerTagDef();
tagDef.setName(entry.getValue().getType());
tagDef.setId(entry.getKey());
ret.put(entry.getKey(), tagDef);
- }
- }
-
- return ret;
- }
-
- static private String getQualifiedNameAttributeName(String entityTypeName) {
- String ret = StringUtils.equals(entityTypeName, ENTITY_TYPE_HIVE_TABLE) ?
- ENTITY_ATTRIBUTE_QUALIFIED_NAME_FOR_HIVE_TABLE : ENTITY_ATTRIBUTE_QUALIFIED_NAME;
-
- return ret;
- }
-
- static private List<String> getQualifiedNameComponents(IReferenceableInstance entity) throws Exception {
- List<String> ret = null;
-
- String qualifiedNameAttributeName = getQualifiedNameAttributeName(entity.getTypeName());
-
- String qualifiedName = getEntityAttribute(entity, qualifiedNameAttributeName, String.class);
-
- ret = getQualifiedNameComponents(entity.getTypeName(), qualifiedName);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("----- Entity-Id:" + entity.getId()._getId());
- LOG.debug("----- Entity-Type-Name:" + entity.getTypeName());
- LOG.debug("----- Entity-Components -----");
- int i = 0;
- for (String value : ret) {
- LOG.debug("----- Index:" + i++ + " Value:" + value);
}
}
- return ret;
- }
-
- static public List<String> getQualifiedNameComponents(String entityTypeName, String qualifiedName) throws Exception {
-
- List<String> ret = null;
-
- String qualifiedNameAttributeName = getQualifiedNameAttributeName(entityTypeName);
-
- if (StringUtils.isBlank(qualifiedName)) {
- throw new Exception("Could not get a valid value for " + qualifiedNameAttributeName + " attribute from entity notification.");
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Received .... " + qualifiedNameAttributeName + "=" + qualifiedName + " for entity type " + entityTypeName);
- }
-
- String components[] = qualifiedName.split(QUALIFIED_NAME_FORMAT_CLUSTER_DELIMITER_STRING);
-
- if (components == null || components.length != 2) {
- throw new Exception("Qualified Name does not contain cluster-name, qualifiedName=" + qualifiedName);
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("name-hierarchy=" + components[0] + ", cluster-name=" + components[1]);
- }
-
- String nameHierarchy[] = components[0].split(QUALIFIED_NAME_FORMAT_DELIMITER_STRING);
-
- int hierarchyLevels = nameHierarchy.length;
-
- ret = new ArrayList<String>();
-
- ret.add(qualifiedName);
- ret.add(components[1]);
-
- for (int i = 0; i < hierarchyLevels; i++) {
- ret.add(nameHierarchy[i]);
- }
return ret;
}
- static private String getServiceName(String clusterName, String entityTypeName) {
- // Parse entityTypeName to get the Apache-component Name
- // Assumption: entityTypeName is <componentName>_<component_specific_type_name>
- // such as hive_table, hadoop_path, hbase_queue, etc.
- String apacheComponents[] = entityTypeName.split("_");
- String apacheComponent = null;
- if (apacheComponents.length > 0) {
- apacheComponent = apacheComponents[0].toLowerCase();
- }
-
- return TagSyncConfig.getServiceName(apacheComponent, clusterName, properties);
- }
-
- static private <T> T getEntityAttribute(IReferenceableInstance entity, String name, Class<T> type) {
- T ret = null;
-
- try {
- Map<String, Object> valueMap = entity.getValuesMap();
- ret = getAttribute(valueMap, name, type);
- } catch (AtlasException exception) {
- LOG.error("Cannot get map of values for entity: " + entity.getId()._getId(), exception);
- }
-
- return ret;
- }
- static private <T> T getAttribute(Map<String, Object> map, String name, Class<T> type) {
- return type.cast(map.get(name));
- }
-
}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/TagAtlasSource.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/TagAtlasSource.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/TagAtlasSource.java
deleted file mode 100644
index 42ba7c7..0000000
--- a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/TagAtlasSource.java
+++ /dev/null
@@ -1,241 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.ranger.tagsync.source.atlas;
-
-import com.google.gson.Gson;
-
-import com.google.inject.Guice;
-import com.google.inject.Injector;
-import com.google.inject.Provider;
-
-import org.apache.atlas.AtlasException;
-import org.apache.commons.collections.MapUtils;
-import org.apache.commons.lang.StringUtils;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-
-import org.apache.atlas.notification.NotificationConsumer;
-import org.apache.atlas.notification.NotificationInterface;
-import org.apache.atlas.notification.NotificationModule;
-import org.apache.atlas.notification.entity.EntityNotification;
-import org.apache.atlas.typesystem.IReferenceableInstance;
-import org.apache.atlas.typesystem.IStruct;
-
-import org.apache.ranger.tagsync.model.TagSink;
-import org.apache.ranger.tagsync.model.TagSource;
-import org.apache.ranger.plugin.util.ServiceTags;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.util.*;
-
-public class TagAtlasSource implements TagSource {
- private static final Log LOG = LogFactory.getLog(TagAtlasSource.class);
-
- public static final String TAGSYNC_ATLAS_PROPERTIES_FILE_NAME = "application.properties";
-
- public static final String TAGSYNC_ATLAS_KAFKA_ENDPOINTS = "atlas.kafka.bootstrap.servers";
- public static final String TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT = "atlas.kafka.zookeeper.connect";
- public static final String TAGSYNC_ATLAS_CONSUMER_GROUP = "atlas.kafka.entities.group.id";
-
- private TagSink tagSink;
- private Properties properties;
- private ConsumerRunnable consumerTask;
-
- @Override
- public boolean initialize(Properties properties) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> TagAtlasSource.initialize()");
- }
-
- boolean ret = true;
-
- if (properties == null || MapUtils.isEmpty(properties)) {
- LOG.error("No properties specified for TagFileSource initialization");
- this.properties = new Properties();
- } else {
- this.properties = properties;
- }
-
- Properties atlasProperties = new Properties();
-
- InputStream inputStream = getClass().getClassLoader().getResourceAsStream(TAGSYNC_ATLAS_PROPERTIES_FILE_NAME);
-
- if (inputStream != null) {
- try {
- atlasProperties.load(inputStream);
- } catch (Exception exception) {
- ret = false;
- LOG.error("Cannot load Atlas application properties file, file-name:" + TAGSYNC_ATLAS_PROPERTIES_FILE_NAME, exception);
- } finally {
- try {
- inputStream.close();
- } catch (IOException ioException) {
- LOG.error("Cannot close Atlas application properties file, file-name:\" + TAGSYNC_ATLAS_PROPERTIES_FILE_NAME", ioException);
- }
- }
- } else {
- ret = false;
- LOG.error("Cannot find Atlas application properties file");
- }
-
- if (ret) {
- if (StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_KAFKA_ENDPOINTS))) {
- ret = false;
- LOG.error("Value of property '" + TAGSYNC_ATLAS_KAFKA_ENDPOINTS + "' is not specified!");
- }
- if (StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT))) {
- ret = false;
- LOG.error("Value of property '" + TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT + "' is not specified!");
- }
- if (StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_CONSUMER_GROUP))) {
- ret = false;
- LOG.error("Value of property '" + TAGSYNC_ATLAS_CONSUMER_GROUP + "' is not specified!");
- }
- }
-
- if (ret) {
- NotificationModule notificationModule = new NotificationModule();
-
- Injector injector = Guice.createInjector(notificationModule);
-
- Provider<NotificationInterface> consumerProvider = injector.getProvider(NotificationInterface.class);
- NotificationInterface notification = consumerProvider.get();
- List<NotificationConsumer<EntityNotification>> iterators = notification.createConsumers(NotificationInterface.NotificationType.ENTITIES, 1);
-
- consumerTask = new ConsumerRunnable(iterators.get(0));
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("<== TagAtlasSource.initialize(), result=" + ret);
- }
- return ret;
- }
-
- @Override
- public void setTagSink(TagSink sink) {
- if (sink == null) {
- LOG.error("Sink is null!!!");
- } else {
- this.tagSink = sink;
- }
- }
-
- @Override
- public Thread start() {
- Thread consumerThread = null;
- if (consumerTask == null) {
- LOG.error("No consumerTask!!!");
- } else {
- consumerThread = new Thread(consumerTask);
- consumerThread.setDaemon(true);
- consumerThread.start();
- }
- return consumerThread;
- }
-
- @Override
- public void updateSink() throws Exception {
- }
-
- @Override
- public boolean isChanged() {
- return true;
- }
-
- // ----- inner class : ConsumerRunnable ------------------------------------
-
- private class ConsumerRunnable implements Runnable {
-
- private final NotificationConsumer<EntityNotification> consumerIterator;
-
- private ConsumerRunnable(NotificationConsumer<EntityNotification> consumerIterator) {
- this.consumerIterator = consumerIterator;
- }
-
- // ----- Runnable --------------------------------------------------------
-
- @Override
- public void run() {
- while (consumerIterator.hasNext()) {
- try {
- EntityNotification notification = consumerIterator.next();
- if (notification != null) {
- printNotification(notification);
- ServiceTags serviceTags = AtlasNotificationMapper.processEntityNotification(notification, properties);
- if (serviceTags == null) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("Did not create ServiceTags structure for notification type:" + notification.getOperationType());
- }
- } else {
- if (LOG.isDebugEnabled()) {
- String serviceTagsJSON = new Gson().toJson(serviceTags);
- LOG.debug("Atlas notification mapped to serviceTags=" + serviceTagsJSON);
- }
-
- try {
- tagSink.uploadServiceTags(serviceTags);
- } catch (Exception exception) {
- LOG.error("uploadServiceTags() failed..", exception);
- }
- }
- }
- } catch(Exception e){
- LOG.error("Exception encountered when processing notification:", e);
- }
- }
- }
-
- public void printNotification(EntityNotification notification) {
- IReferenceableInstance entity = notification.getEntity();
- if (LOG.isDebugEnabled()) {
- try {
- LOG.debug("Notification-Type: " + notification.getOperationType());
- LOG.debug("Entity-Id: " + entity.getId()._getId());
- LOG.debug("Entity-Type: " + entity.getTypeName());
-
- LOG.debug("----------- Entity Values ----------");
-
-
- for (Map.Entry<String, Object> entry : entity.getValuesMap().entrySet()) {
- LOG.debug(" Name:" + entry.getKey());
- Object value = entry.getValue();
- LOG.debug(" Value:" + value);
- }
-
- LOG.debug("----------- Entity Traits ----------");
-
- List<IStruct> traits = notification.getAllTraits();
-
- for (IStruct trait : traits) {
- LOG.debug(" Trait-Type-Name:" + trait.getTypeName());
- Map<String, Object> traitValues = trait.getValuesMap();
- for (Map.Entry<String, Object> valueEntry : traitValues.entrySet()) {
- LOG.debug(" Trait-Value-Name:" + valueEntry.getKey());
- LOG.debug(" Trait-Value:" + valueEntry.getValue());
- }
- }
- } catch (AtlasException exception) {
- LOG.error("Cannot print notification - ", exception);
- }
- }
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java
new file mode 100644
index 0000000..c8ed948
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTTagSource.java
@@ -0,0 +1,143 @@
+/*
+ * 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.ranger.tagsync.source.atlasrest;
+
+import com.google.gson.Gson;
+
+import com.google.gson.GsonBuilder;
+
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.ranger.tagsync.model.AbstractTagSource;
+import org.apache.ranger.plugin.util.ServiceTags;
+import org.apache.ranger.tagsync.process.TagSyncConfig;
+import org.apache.ranger.tagsync.source.atlas.AtlasEntityWithTraits;
+import org.apache.ranger.tagsync.source.atlas.AtlasNotificationMapper;
+import org.apache.ranger.tagsync.source.atlas.AtlasResourceMapperUtil;
+
+import java.util.*;
+
+public class AtlasRESTTagSource extends AbstractTagSource implements Runnable {
+ private static final Log LOG = LogFactory.getLog(AtlasRESTTagSource.class);
+
+ private String atlasEndpoint;
+ private long sleepTimeBetweenCycleInMillis;
+
+ @Override
+ public boolean initialize(Properties properties) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> AtlasRESTTagSource.initialize()");
+ }
+
+ boolean ret = AtlasResourceMapperUtil.initializeAtlasResourceMappers(properties);
+
+ atlasEndpoint = TagSyncConfig.getAtlasEndpoint(properties);
+ sleepTimeBetweenCycleInMillis = TagSyncConfig.getTagSourceAtlasDownloadIntervalInMillis(properties);
+
+ if (StringUtils.isEmpty(atlasEndpoint)) {
+ LOG.info("No AtlasEndpoint specified, Initial download of Atlas-entities cannot be done.");
+ ret = false;
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== AtlasRESTTagSource.initialize(), result=" + ret);
+ }
+
+ return ret;
+ }
+
+ @Override
+ public boolean start() {
+
+ Thread atlasRESTInvokerThread = new Thread(this);
+ atlasRESTInvokerThread.setDaemon(true);
+ atlasRESTInvokerThread.start();
+
+ return atlasRESTInvokerThread != null;
+ }
+
+ @Override
+ public void run() {
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> AtlasRESTTagSource.run()");
+ }
+
+ while (!shutdown) {
+
+ synchUp();
+
+ LOG.debug("Sleeping for [" + sleepTimeBetweenCycleInMillis + "] milliSeconds");
+
+ try {
+
+ Thread.sleep(sleepTimeBetweenCycleInMillis);
+
+ } catch (InterruptedException exception) {
+ LOG.error("Failed to wait for [" + sleepTimeBetweenCycleInMillis + "] milliseconds before checking for update to tagFileSource", exception);
+ }
+ }
+ LOG.info("Shutting down the Tag-Atlasrest-source thread");
+ }
+
+ @Override
+ public boolean isChanged() {
+ return true;
+ }
+
+ @Override
+ public void synchUp() {
+
+ AtlasRESTUtil atlasRESTUtil = new AtlasRESTUtil(atlasEndpoint);
+
+ List<AtlasEntityWithTraits> atlasEntitiesWithTraits = atlasRESTUtil.getEntitiesWithTraits();
+
+ if (CollectionUtils.isNotEmpty(atlasEntitiesWithTraits)) {
+ if (LOG.isDebugEnabled()) {
+ for (AtlasEntityWithTraits element : atlasEntitiesWithTraits) {
+ LOG.debug(element);
+ }
+ }
+
+ Map<String, ServiceTags> serviceTagsMap = AtlasNotificationMapper.processEntitiesWithTraits(atlasEntitiesWithTraits);
+
+ if (MapUtils.isNotEmpty(serviceTagsMap)) {
+ for (Map.Entry<String, ServiceTags> entry : serviceTagsMap.entrySet()) {
+ if (LOG.isDebugEnabled()) {
+ Gson gsonBuilder = new GsonBuilder().setDateFormat("yyyyMMdd-HH:mm:ss.SSS-Z")
+ .setPrettyPrinting()
+ .create();
+ String serviceTagsString = gsonBuilder.toJson(entry.getValue());
+
+ LOG.debug("serviceTags=" + serviceTagsString);
+ }
+ updateSink(entry.getValue());
+ }
+ }
+ }
+
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTUtil.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTUtil.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTUtil.java
new file mode 100644
index 0000000..7f4676a
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlasrest/AtlasRESTUtil.java
@@ -0,0 +1,264 @@
+/**
+ * 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.ranger.tagsync.source.atlasrest;
+
+import com.google.gson.Gson;
+import com.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.WebResource;
+import org.apache.atlas.typesystem.IReferenceableInstance;
+import org.apache.atlas.typesystem.IStruct;
+import org.apache.atlas.typesystem.Struct;
+import org.apache.atlas.typesystem.json.InstanceSerialization;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.collections.MapUtils;
+import org.apache.log4j.Logger;
+import org.apache.ranger.admin.client.datatype.RESTResponse;
+import org.apache.ranger.plugin.util.RangerRESTClient;
+import org.apache.ranger.tagsync.source.atlas.AtlasEntityWithTraits;
+import org.apache.ranger.tagsync.source.atlas.AtlasResourceMapperUtil;
+
+import java.util.*;
+
+@SuppressWarnings("unchecked")
+public class AtlasRESTUtil {
+ private static final Logger LOG = Logger.getLogger(AtlasRESTUtil.class);
+
+ private static final String REST_MIME_TYPE_JSON = "application/json" ;
+ private static final String API_ATLAS_TYPES = "api/atlas/types";
+ private static final String API_ATLAS_ENTITIES = "api/atlas/entities?type=";
+ private static final String API_ATLAS_ENTITY = "api/atlas/entities/";
+ private static final String API_ATLAS_TYPE = "api/atlas/types/";
+
+ private static final String RESULTS_ATTRIBUTE = "results";
+ private static final String DEFINITION_ATTRIBUTE = "definition";
+ private static final String VALUES_ATTRIBUTE = "values";
+ private static final String TRAITS_ATTRIBUTE = "traits";
+ private static final String TYPE_NAME_ATTRIBUTE = "typeName";
+ private static final String TRAIT_TYPES_ATTRIBUTE = "traitTypes";
+ private static final String SUPER_TYPES_ATTRIBUTE = "superTypes";
+ private static final String ATTRIBUTE_DEFINITIONS_ATTRIBUTE = "attributeDefinitions";
+ private static final String NAME_ATTRIBUTE = "name";
+
+ private final Gson gson = new Gson();
+
+ private RangerRESTClient atlasRESTClient;
+
+ public AtlasRESTUtil(String atlasEndpoint) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> AtlasRESTUtil(" + atlasEndpoint + ")");
+ }
+
+ if (!atlasEndpoint.endsWith("/")) {
+ atlasEndpoint += "/";
+ }
+
+ // This uses RangerRESTClient to invoke REST APIs on Atlas. It will work only if scheme of URL is http
+ atlasRESTClient = new RangerRESTClient();
+ atlasRESTClient.setUrl(atlasEndpoint);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== AtlasRESTUtil(" + atlasEndpoint + ")");
+ }
+ }
+
+ public List<AtlasEntityWithTraits> getEntitiesWithTraits() {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> getEntriesWithTraits()");
+ }
+
+ List<AtlasEntityWithTraits> ret = new ArrayList<AtlasEntityWithTraits>();
+
+ Map<String, Object> typesResponse = atlasAPI(API_ATLAS_TYPES);
+
+ List<String> types = getAttribute(typesResponse, RESULTS_ATTRIBUTE, List.class);
+
+ if (CollectionUtils.isNotEmpty(types)) {
+
+ for (String type : types) {
+
+ if (!AtlasResourceMapperUtil.isEntityTypeHandled(type)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Not fetching Atlas entities of type:" + type);
+ }
+ continue;
+ }
+
+ Map<String, Object> entitiesResponse = atlasAPI(API_ATLAS_ENTITIES + type);
+
+ List<String> guids = getAttribute(entitiesResponse, RESULTS_ATTRIBUTE, List.class);
+
+ if (CollectionUtils.isEmpty(guids)) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("No Atlas entities for type:" + type);
+ }
+ continue;
+ }
+
+ for (String guid : guids) {
+
+ Map<String, Object> entityResponse = atlasAPI(API_ATLAS_ENTITY + guid);
+
+ if (MapUtils.isNotEmpty(entityResponse) && entityResponse.containsKey(DEFINITION_ATTRIBUTE)) {
+
+ Map<String, Object> definition = getAttribute(entityResponse, DEFINITION_ATTRIBUTE, Map.class);
+ Map<String, Object> traitsAttribute = getAttribute(definition, TRAITS_ATTRIBUTE, Map.class);
+
+ if (MapUtils.isNotEmpty(traitsAttribute)) {
+
+ List<IStruct> allTraits = new LinkedList<>();
+
+ for (Map.Entry<String, Object> entry : traitsAttribute.entrySet()) {
+
+ Map<String, Object> trait = (Map<String, Object>) entry.getValue();
+
+ Map<String, Object> traitValues = getAttribute(trait, VALUES_ATTRIBUTE, Map.class);
+ String traitTypeName = getAttribute(trait, TYPE_NAME_ATTRIBUTE, String.class);
+
+ List<IStruct> superTypes = getTraitSuperTypes(getTraitType(traitTypeName), traitValues);
+
+ Struct trait1 = new Struct(traitTypeName, traitValues);
+
+ allTraits.add(trait1);
+ allTraits.addAll(superTypes);
+ }
+
+ IReferenceableInstance entity = InstanceSerialization.fromJsonReferenceable(gson.toJson(definition), true);
+
+ if (entity != null) {
+ AtlasEntityWithTraits entityWithTraits = new AtlasEntityWithTraits(entity, allTraits);
+ ret.add(entityWithTraits);
+ } else {
+ if (LOG.isInfoEnabled()) {
+ LOG.info("Could not create Atlas entity from its definition, type=" + type + ", guid=" + guid);
+ }
+ }
+
+ }
+ }
+ }
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== getEntriesWithTraits()");
+ }
+ }
+
+ return ret;
+ }
+
+ private Map<String, Object> getTraitType(String traitName) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> getTraitType(" + traitName + ")");
+ }
+ Map<String, Object> ret = null;
+
+ Map<String, Object> typeResponse = atlasAPI(API_ATLAS_TYPE + traitName);
+
+ if (typeResponse.containsKey(DEFINITION_ATTRIBUTE)) {
+
+ Map<String, Object> definition = getAttribute(typeResponse, DEFINITION_ATTRIBUTE, Map.class);
+
+ List traitTypes = getAttribute(definition, TRAIT_TYPES_ATTRIBUTE, List.class);
+
+ if (traitTypes.size() > 0) {
+ ret = (Map<String, Object>) traitTypes.get(0);
+ }
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== getTraitType(" + traitName + ")");
+ }
+ return ret;
+ }
+
+ private List<IStruct> getTraitSuperTypes(Map<String, Object> traitType, Map<String, Object> values) {
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> getTraitSuperTypes()");
+ }
+ List<IStruct> ret = new LinkedList<>();
+
+ if (traitType != null) {
+
+ List<String> superTypeNames = getAttribute(traitType, SUPER_TYPES_ATTRIBUTE, List.class);
+
+ for (String superTypeName : superTypeNames) {
+
+ Map<String, Object> superTraitType = getTraitType(superTypeName);
+
+ if (superTraitType != null) {
+ List<Map<String, Object>> attributeDefinitions = (List) superTraitType.get(ATTRIBUTE_DEFINITIONS_ATTRIBUTE);
+
+ Map<String, Object> superTypeValues = new HashMap<>();
+ for (Map<String, Object> attributeDefinition : attributeDefinitions) {
+
+ String attributeName = attributeDefinition.get(NAME_ATTRIBUTE).toString();
+ if (values.containsKey(attributeName)) {
+ superTypeValues.put(attributeName, values.get(attributeName));
+ }
+ }
+
+ List<IStruct> superTraits = getTraitSuperTypes(getTraitType(superTypeName), values);
+
+ Struct superTrait = new Struct(superTypeName, superTypeValues);
+
+ ret.add(superTrait);
+ ret.addAll(superTraits);
+ }
+ }
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== getTraitSuperTypes()");
+ }
+ return ret;
+ }
+
+ private Map<String, Object> atlasAPI(String endpoint) {
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> atlasAPI(" + endpoint + ")");
+ }
+ Map<String, Object> ret = new HashMap<String, Object>();
+
+ try {
+ WebResource webResource = atlasRESTClient.getResource(endpoint);
+ ClientResponse response = webResource.accept(REST_MIME_TYPE_JSON).type(REST_MIME_TYPE_JSON).get(ClientResponse.class);
+
+ if (response != null && response.getStatus() == 200) {
+ ret = response.getEntity(ret.getClass());
+ } else {
+ RESTResponse resp = RESTResponse.fromClientResponse(response);
+ LOG.error("Error getting atlas data request=" + webResource.toString()
+ + ", response=" + resp.toString());
+ }
+ } catch (Exception exception) {
+ LOG.error("Exception when fetching Atlas objects.", exception);
+ ret = null;
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== atlasAPI(" + endpoint + ")");
+ }
+ return ret;
+ }
+
+ private <T> T getAttribute(Map<String, Object> map, String name, Class<T> type) {
+ return MapUtils.isNotEmpty(map) ? type.cast(map.get(name)) : null;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/source/file/FileTagSource.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/file/FileTagSource.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/file/FileTagSource.java
new file mode 100644
index 0000000..43eb3b5
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/file/FileTagSource.java
@@ -0,0 +1,278 @@
+/*
+ * 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.ranger.tagsync.source.file;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.ranger.tagsync.model.AbstractTagSource;
+import org.apache.ranger.plugin.util.ServiceTags;
+import org.apache.ranger.tagsync.process.TagSyncConfig;
+
+import java.io.*;
+import java.net.MalformedURLException;
+import java.net.URL;
+import java.nio.charset.Charset;
+import java.util.Date;
+import java.util.Properties;
+
+public class FileTagSource extends AbstractTagSource implements Runnable {
+ private static final Log LOG = LogFactory.getLog(FileTagSource.class);
+
+ private String serviceTagsFileName;
+ private URL serviceTagsFileURL;
+ private long lastModifiedTimeInMillis = -1L;
+
+ private Gson gsonBuilder;
+ private Properties properties;
+ private long fileModTimeCheckIntervalInMs;
+
+ @Override
+ public boolean initialize(Properties props) {
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> FileTagSource.initialize()");
+ }
+
+ if (props == null || MapUtils.isEmpty(props)) {
+ LOG.error("No properties specified for FileTagSource initialization");
+ this.properties = new Properties();
+ } else {
+ this.properties = props;
+ }
+
+ gsonBuilder = new GsonBuilder().setDateFormat("yyyyMMdd-HH:mm:ss.SSS-Z").setPrettyPrinting().create();
+
+ boolean ret = true;
+
+ serviceTagsFileName = TagSyncConfig.getTagSourceFileName(properties);
+
+ if (StringUtils.isBlank(serviceTagsFileName)) {
+ ret = false;
+ LOG.error("value of property 'ranger.tagsync.source.impl.class' is file and no value specified for property 'ranger.tagsync.filesource.filename'!");
+ }
+
+ if (ret) {
+
+ fileModTimeCheckIntervalInMs = TagSyncConfig.getTagSourceFileModTimeCheckIntervalInMillis(properties);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Provided serviceTagsFileName=" + serviceTagsFileName);
+ LOG.debug("'ranger.tagsync.filesource.modtime.check.interval':" + fileModTimeCheckIntervalInMs + "ms");
+ }
+
+ InputStream serviceTagsFileStream = null;
+
+ File f = new File(serviceTagsFileName);
+
+ if (f.exists() && f.isFile() && f.canRead()) {
+ try {
+ serviceTagsFileStream = new FileInputStream(f);
+ serviceTagsFileURL = f.toURI().toURL();
+ } catch (FileNotFoundException exception) {
+ LOG.error("Error processing input file:" + serviceTagsFileName + " or no privilege for reading file " + serviceTagsFileName, exception);
+ } catch (MalformedURLException malformedException) {
+ LOG.error("Error processing input file:" + serviceTagsFileName + " cannot be converted to URL " + serviceTagsFileName, malformedException);
+ }
+ } else {
+
+ URL fileURL = getClass().getResource(serviceTagsFileName);
+ if (fileURL == null) {
+ if (!serviceTagsFileName.startsWith("/")) {
+ fileURL = getClass().getResource("/" + serviceTagsFileName);
+ }
+ }
+
+ if (fileURL == null) {
+ fileURL = ClassLoader.getSystemClassLoader().getResource(serviceTagsFileName);
+ if (fileURL == null) {
+ if (!serviceTagsFileName.startsWith("/")) {
+ fileURL = ClassLoader.getSystemClassLoader().getResource("/" + serviceTagsFileName);
+ }
+ }
+ }
+
+ if (fileURL != null) {
+
+ try {
+ serviceTagsFileStream = fileURL.openStream();
+ serviceTagsFileURL = fileURL;
+ } catch (Exception exception) {
+ LOG.error(serviceTagsFileName + " is not a file", exception);
+ }
+ } else {
+ LOG.warn("Error processing input file: URL not found for " + serviceTagsFileName + " or no privilege for reading file " + serviceTagsFileName);
+ }
+ }
+
+ if (serviceTagsFileStream != null) {
+ try {
+ serviceTagsFileStream.close();
+ } catch (Exception e) {
+ // Ignore
+ }
+ }
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== FileTagSource.initialize(): sourceFileName=" + serviceTagsFileName + ", result=" + ret);
+ }
+
+ return ret;
+ }
+
+ @Override
+ public boolean start() {
+
+ Thread fileMonitoringThread = new Thread(this);
+ fileMonitoringThread.setDaemon(true);
+ fileMonitoringThread.start();
+
+ return fileMonitoringThread != null;
+ }
+
+ @Override
+ public void run() {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> FileTagSource.run()");
+ }
+
+ while (!shutdown) {
+
+ try {
+ synchUp();
+
+ LOG.debug("Sleeping for [" + fileModTimeCheckIntervalInMs + "] milliSeconds");
+
+ Thread.sleep(fileModTimeCheckIntervalInMs);
+ }
+ catch (InterruptedException e) {
+ LOG.error("Failed to wait for [" + fileModTimeCheckIntervalInMs + "] milliseconds before checking for update to tagFileSource", e);
+ }
+ catch (Throwable t) {
+ LOG.error("tag-sync thread got an error", t);
+ }
+ }
+
+ LOG.info("Shutting down the Tag-file-source thread");
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== FileTagSource.run()");
+ }
+ }
+
+ @Override
+ public boolean isChanged() {
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> FileTagSource.isChanged()");
+ }
+ boolean ret = false;
+
+ long modificationTime = getModificationTime();
+
+ if (modificationTime > lastModifiedTimeInMillis) {
+ if (LOG.isDebugEnabled()) {
+ Date modifiedDate = new Date(modificationTime);
+ Date lastModifiedDate = new Date(lastModifiedTimeInMillis);
+ LOG.debug("File modified at " + modifiedDate + "last-modified at " + lastModifiedDate);
+ }
+ lastModifiedTimeInMillis = modificationTime;
+ ret = true;
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== FileTagSource.isChanged(): result=" + ret);
+ }
+ return ret;
+ }
+
+ @Override
+ public void synchUp() {
+ if (isChanged()) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Begin: update tags from source==>sink");
+ }
+
+ ServiceTags serviceTags = readFromFile();
+ updateSink(serviceTags);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("End: update tags from source==>sink");
+ }
+
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("FileTagSource: no change found for synchronization.");
+ }
+ }
+ }
+ private ServiceTags readFromFile() {
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> FileTagSource.readFromFile(): sourceFileName=" + serviceTagsFileName);
+ }
+
+ ServiceTags ret = null;
+
+ if (serviceTagsFileURL != null) {
+ try (
+ InputStream serviceTagsFileStream = serviceTagsFileURL.openStream();
+ Reader reader = new InputStreamReader(serviceTagsFileStream, Charset.forName("UTF-8"))
+ ) {
+
+ ret = gsonBuilder.fromJson(reader, ServiceTags.class);
+
+ } catch (IOException e) {
+ LOG.warn("Error processing input file: or no privilege for reading file " + serviceTagsFileName, e);
+ }
+ } else {
+ LOG.error("Error reading file: " + serviceTagsFileName);
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== FileTagSource.readFromFile(): sourceFileName=" + serviceTagsFileName);
+ }
+
+ return ret;
+ }
+
+ private long getModificationTime() {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> FileTagSource.getLastModificationTime(): sourceFileName=" + serviceTagsFileName);
+ }
+ long ret = 0L;
+
+ File sourceFile = new File(serviceTagsFileName);
+
+ if (sourceFile.exists() && sourceFile.isFile() && sourceFile.canRead()) {
+ ret = sourceFile.lastModified();
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== FileTagSource.lastModificationTime(): serviceTagsFileName=" + serviceTagsFileName + " result=" + new Date(ret));
+ }
+
+ return ret;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/source/file/TagFileSource.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/file/TagFileSource.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/file/TagFileSource.java
deleted file mode 100644
index 92f24b2..0000000
--- a/tagsync/src/main/java/org/apache/ranger/tagsync/source/file/TagFileSource.java
+++ /dev/null
@@ -1,281 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.ranger.tagsync.source.file;
-
-import com.google.gson.Gson;
-import com.google.gson.GsonBuilder;
-import org.apache.commons.collections.MapUtils;
-import org.apache.commons.lang.StringUtils;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.ranger.tagsync.model.TagSink;
-import org.apache.ranger.tagsync.model.TagSource;
-import org.apache.ranger.plugin.util.ServiceTags;
-import org.apache.ranger.tagsync.process.TagSyncConfig;
-
-import java.io.*;
-import java.util.Date;
-import java.util.Properties;
-
-public class TagFileSource implements TagSource, Runnable {
- private static final Log LOG = LogFactory.getLog(TagFileSource.class);
-
- private String sourceFileName;
- private long lastModifiedTimeInMillis = 0L;
-
- private Gson gson;
- private TagSink tagSink;
- private Properties properties;
-
- @Override
- public boolean initialize(Properties props) {
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> TagFileSource.initialize()");
- }
-
- if (props == null || MapUtils.isEmpty(props)) {
- LOG.error("No properties specified for TagFileSource initialization");
- this.properties = new Properties();
- } else {
- this.properties = props;
- }
-
- boolean ret = true;
-
- if (StringUtils.isBlank(TagSyncConfig.getTagSourceFileName(properties))) {
- ret = false;
- LOG.error("value of property 'ranger.tagsync.source.impl.class' is file and no value specified for property 'ranger.tagsync.filesource.filename'!");
- }
-
- if (ret) {
-
- long fileModTimeCheckIntervalInMs = TagSyncConfig.getTagSourceFileModTimeCheckIntervalInMillis(properties);
-
- if (fileModTimeCheckIntervalInMs <= 0L) {
- LOG.info("'ranger.tagsync.filesource.modtime.check.interval' is zero or negative! 'ranger.tagsync.filesource.modtime.check.interval'=" + fileModTimeCheckIntervalInMs + "ms");
- LOG.info("Setting 'ranger.tagsync.filesource.modtime.check.interval' to 60 seconds");
- fileModTimeCheckIntervalInMs = 60*1000;
- } else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("'ranger.tagsync.filesource.modtime.check.interval':" + fileModTimeCheckIntervalInMs + "ms");
- }
- }
- sourceFileName = TagSyncConfig.getTagSourceFileName(properties);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("Provided sourceFileName=" + sourceFileName);
- }
-
- String realFileName = TagSyncConfig.getResourceFileName(sourceFileName);
- if (realFileName != null) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Real sourceFileName=" + realFileName);
- }
- sourceFileName = realFileName;
- } else {
- LOG.error(sourceFileName + " is not a file or is not readable");
- ret = false;
- }
- }
-
- if (ret) {
- try {
- gson = new GsonBuilder().setDateFormat("yyyyMMdd-HH:mm:ss.SSS-Z").setPrettyPrinting().create();
- } catch (Throwable excp) {
- LOG.fatal("failed to create GsonBuilder object", excp);
- ret = false;
- }
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("<== TagFileSource.initialize(): sourceFileName=" + sourceFileName + ", result=" + ret);
- }
-
- return ret;
- }
-
- @Override
- public void setTagSink(TagSink sink) {
- if (sink == null) {
- LOG.error("Sink is null!!!");
- } else {
- this.tagSink = sink;
- }
- }
-
- @Override
- public Thread start() {
-
- Thread fileMonitoringThread = null;
-
- fileMonitoringThread = new Thread(this);
- fileMonitoringThread.setDaemon(true);
- fileMonitoringThread.start();
-
- return fileMonitoringThread;
- }
-
- @Override
- public void run() {
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> TagFileSource.run()");
- }
- long sleepTimeBetweenCycleInMillis = TagSyncConfig.getTagSourceFileModTimeCheckIntervalInMillis(properties);
- boolean shutdownFlag = false;
-
- while (!shutdownFlag) {
-
- try {
- if (isChanged()) {
- LOG.info("Begin: update tags from source==>sink");
- if (TagSyncConfig.isTagSyncEnabled(properties)) {
- updateSink();
- LOG.info("End: update tags from source==>sink");
- } else {
- LOG.info("Tag-sync is not enabled.");
- }
- } else {
- LOG.debug("TagFileSource: no change found for synchronization.");
- }
-
- LOG.debug("Sleeping for [" + sleepTimeBetweenCycleInMillis + "] milliSeconds");
-
- Thread.sleep(sleepTimeBetweenCycleInMillis);
- }
- catch (InterruptedException e) {
- LOG.error("Failed to wait for [" + sleepTimeBetweenCycleInMillis + "] milliseconds before checking for update to tagFileSource", e);
- shutdownFlag = true;
- }
- catch (Throwable t) {
- LOG.error("tag-sync thread got an error", t);
- }
- }
-
- LOG.info("Shutting down the Tag-file-source thread");
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("<== TagFileSource.run()");
- }
- }
-
- @Override
- public void updateSink() throws Exception {
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> TagFileSource.updateSink()");
- }
- ServiceTags serviceTags = readFromFile();
-
- if (serviceTags != null) {
- tagSink.uploadServiceTags(serviceTags);
- } else {
- LOG.error("Could not read ServiceTags from file");
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("<== TagFileSource.updateSink()");
- }
- }
-
- @Override
- public boolean isChanged() {
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> TagFileSource.isChanged()");
- }
- boolean ret = false;
-
- long modificationTime = getModificationTime();
-
- if (modificationTime > lastModifiedTimeInMillis) {
- if (LOG.isDebugEnabled()) {
- Date modifiedDate = new Date(modificationTime);
- Date lastModifiedDate = new Date(lastModifiedTimeInMillis);
- LOG.debug("File modified at " + modifiedDate + "last-modified at " + lastModifiedDate);
- }
- lastModifiedTimeInMillis = modificationTime;
- ret = true;
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("<== TagFileSource.isChanged(): result=" + ret);
- }
- return ret;
- }
-
- private ServiceTags readFromFile() {
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> TagFileSource.readFromFile(): sourceFileName=" + sourceFileName);
- }
-
- ServiceTags ret = null;
-
- Reader reader = null;
- try {
-
- reader = new InputStreamReader(TagSyncConfig.getFileInputStream(sourceFileName));
-
- ret = gson.fromJson(reader, ServiceTags.class);
-
- }
- catch (FileNotFoundException exception) {
- LOG.warn("Tag-source file does not exist or not readble '" + sourceFileName + "'");
- }
- catch (Exception excp) {
- LOG.error("failed to load service-tags from Tag-source file " + sourceFileName, excp);
- }
- finally {
- if (reader != null) {
- try {
- reader.close();
- } catch (Exception excp) {
- LOG.error("error while closing opened Tag-source file " + sourceFileName, excp);
- }
- }
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("<== TagFileSource.readFromFile(): sourceFileName=" + sourceFileName);
- }
-
- return ret;
- }
-
- private long getModificationTime() {
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> TagFileSource.getLastModificationTime(): sourceFileName=" + sourceFileName);
- }
- long ret = 0L;
-
- File sourceFile = new File(sourceFileName);
-
- if (sourceFile.exists() && sourceFile.isFile() && sourceFile.canRead()) {
- ret = sourceFile.lastModified();
- } else {
- ret = new Date().getTime();
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("<== TagFileSource.lastModificationTime(): sourceFileName=" + sourceFileName + " result=" + new Date(ret));
- }
-
- return ret;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/resources/etc/ranger/data/tags.json
----------------------------------------------------------------------
diff --git a/tagsync/src/main/resources/etc/ranger/data/tags.json b/tagsync/src/main/resources/etc/ranger/data/tags.json
index 274cf69..b4cd736 100644
--- a/tagsync/src/main/resources/etc/ranger/data/tags.json
+++ b/tagsync/src/main/resources/etc/ranger/data/tags.json
@@ -1,9 +1,7 @@
{
"op":"add_or_update",
- "tagModel":"shared",
+ "tagModel": "resource_private",
"serviceName": "cl1_hive",
- "tagVersion": 24,
- "tagUpdateTime": "20150901-20:03:17.000-+0000",
"tagDefinitions": {
"1": {
"name": "EXPIRES_ON",
@@ -15,12 +13,7 @@
}
],
"id": 1,
- "guid": "1441137512654_323_77",
- "isEnabled": true,
- "createdBy": "Admin",
- "updatedBy": "Admin",
- "createTime": "20150901-19:58:33.000-+0000",
- "updateTime": "20150901-19:58:33.000-+0000"
+ "guid": "tagdefinition-1-guid"
}
},
"tags": {
@@ -30,12 +23,7 @@
"expiry_date": "2014/12/31"
},
"id": 1,
- "guid": "1441137512698_844_80",
- "isEnabled": true,
- "createdBy": "Admin",
- "updatedBy": "Admin",
- "createTime": "20150901-19:58:33.000-+0000",
- "updateTime": "20150901-20:03:17.000-+0000"
+ "guid": "tag-1-guid"
}
},
"serviceResources": [
@@ -57,14 +45,8 @@
"isRecursive": false
}
},
- "resourceSignature": "c1114679b35a65d28e0dca4fdffc27d6",
"id": 1,
- "guid": "1441137512756_887_83",
- "isEnabled": true,
- "createdBy": "Admin",
- "updatedBy": "Admin",
- "createTime": "20150901-19:58:33.000-+0000",
- "updateTime": "20150901-19:58:33.000-+0000"
+ "guid": "serviceresource-1-guid"
}
],
"resourceToTagIds": {
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/resources/ranger-tagsync-default.xml
----------------------------------------------------------------------
diff --git a/tagsync/src/main/resources/ranger-tagsync-default.xml b/tagsync/src/main/resources/ranger-tagsync-default.xml
index b917d29..3784df8 100644
--- a/tagsync/src/main/resources/ranger-tagsync-default.xml
+++ b/tagsync/src/main/resources/ranger-tagsync-default.xml
@@ -55,4 +55,9 @@
<value>tagadmin</value>
<description></description>
</property>
+ <property>
+ <name>ranger.tagsync.atlasrestsource.download.interval</name>
+ <value>600000</value>
+ <description></description>
+ </property>
</configuration>
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/test/java/org/apache/ranger/tagsync/process/TestTagSynchronizer.java
----------------------------------------------------------------------
diff --git a/tagsync/src/test/java/org/apache/ranger/tagsync/process/TestTagSynchronizer.java b/tagsync/src/test/java/org/apache/ranger/tagsync/process/TestTagSynchronizer.java
index 59d521c..43c22d0 100644
--- a/tagsync/src/test/java/org/apache/ranger/tagsync/process/TestTagSynchronizer.java
+++ b/tagsync/src/test/java/org/apache/ranger/tagsync/process/TestTagSynchronizer.java
@@ -20,12 +20,7 @@
package org.apache.ranger.tagsync.process;
-import org.apache.commons.lang.StringUtils;
-import org.apache.ranger.tagsync.model.TagSource;
-import org.apache.ranger.tagsync.process.TagSyncConfig;
-import org.apache.ranger.tagsync.process.TagSynchronizer;
-import org.apache.ranger.tagsync.source.atlas.AtlasNotificationMapper;
-import org.apache.ranger.tagsync.source.atlas.TagAtlasSource;
+import org.apache.ranger.tagsync.source.atlas.AtlasHiveResourceMapper;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -34,8 +29,6 @@ import java.io.*;
import java.util.List;
import java.util.Properties;
-import static org.junit.Assert.*;
-
public class TestTagSynchronizer {
@@ -79,7 +72,7 @@ public class TestTagSynchronizer {
@Test
public void testTagDownload() {
- boolean initDone = false;
+ boolean initDone = true;
/* For tagSynchronizer.initialize() to succeed, edit ranger-tagsync-site.xml file to contain correct
values of the following properties:
@@ -97,10 +90,12 @@ public class TestTagSynchronizer {
*/
- //initDone = tagSynchronizer.initialize();
+// initDone = tagSynchronizer.initialize(null);
System.out.println("TagSynchronizer initialization result=" + initDone);
+ assert(initDone);
+
System.out.println("Exiting testTagDownload()");
}
@@ -108,14 +103,15 @@ public class TestTagSynchronizer {
public void testQualifiedNames() {
List<String> components;
+ AtlasHiveResourceMapper hiveResourceBuilder = new AtlasHiveResourceMapper();
try {
- components = AtlasNotificationMapper.getQualifiedNameComponents("hive_db", "database@cluster");
+ components = hiveResourceBuilder.getQualifiedNameComponents("hive_db", "database@cluster");
printComponents(components);
- components = AtlasNotificationMapper.getQualifiedNameComponents("hive_table", "database.table@cluster");
+ components = hiveResourceBuilder.getQualifiedNameComponents("hive_table", "database.table@cluster");
printComponents(components);
- components = AtlasNotificationMapper.getQualifiedNameComponents("hive_column", "database.table.column@cluster");
+ components = hiveResourceBuilder.getQualifiedNameComponents("hive_column", "database.table.column@cluster");
printComponents(components);
assert(true);
[2/2] incubator-ranger git commit: RANGER-807: TagSync should support
periodic full sync with Apache Atlas
Posted by ma...@apache.org.
RANGER-807: TagSync should support periodic full sync with Apache Atlas
Signed-off-by: Madhan Neethiraj <ma...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-ranger/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-ranger/commit/5b9c094f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-ranger/tree/5b9c094f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-ranger/diff/5b9c094f
Branch: refs/heads/master
Commit: 5b9c094ff501ec591f6334bcea10ce37a2163711
Parents: fec8460
Author: Abhay Kulkarni <ak...@hortonworks.com>
Authored: Tue Dec 15 19:08:02 2015 -0800
Committer: Madhan Neethiraj <ma...@apache.org>
Committed: Sat Jan 9 23:59:10 2016 -0800
----------------------------------------------------------------------
.../ranger/plugin/util/RangerRESTClient.java | 3 +-
.../apache/ranger/plugin/util/ServiceTags.java | 27 +-
.../conf/templates/installprop2xml.properties | 6 +-
.../conf/templates/ranger-tagsync-template.xml | 13 +-
tagsync/scripts/install.properties | 12 +-
tagsync/scripts/ranger-tagsync-services.sh | 6 +-
tagsync/scripts/ranger-tagsync-sync.sh | 66 ++++
tagsync/scripts/setup.py | 19 +-
.../ranger/tagsync/model/AbstractTagSource.java | 65 ++++
.../apache/ranger/tagsync/model/TagSource.java | 6 +-
.../ranger/tagsync/process/TagSyncConfig.java | 68 ++--
.../ranger/tagsync/process/TagSynchronizer.java | 193 +++++++----
.../tagsync/sink/tagadmin/TagAdminRESTSink.java | 162 +++++++++
.../tagsync/sink/tagadmin/TagRESTSink.java | 160 ---------
.../source/Atlas/AtlasHiveResourceMapper.java | 203 ++++++++++++
.../source/Atlas/AtlasResourceMapper.java | 74 +++++
.../source/Atlas/AtlasResourceMapperUtil.java | 124 +++++++
.../tagsync/source/Atlas/AtlasTagSource.java | 197 +++++++++++
.../source/atlas/AtlasEntityWithTraits.java | 96 ++++++
.../source/atlas/AtlasNotificationMapper.java | 327 ++++++-------------
.../tagsync/source/atlas/TagAtlasSource.java | 241 --------------
.../source/atlasrest/AtlasRESTTagSource.java | 143 ++++++++
.../tagsync/source/atlasrest/AtlasRESTUtil.java | 264 +++++++++++++++
.../tagsync/source/file/FileTagSource.java | 278 ++++++++++++++++
.../tagsync/source/file/TagFileSource.java | 281 ----------------
.../main/resources/etc/ranger/data/tags.json | 26 +-
.../main/resources/ranger-tagsync-default.xml | 5 +
.../tagsync/process/TestTagSynchronizer.java | 22 +-
28 files changed, 2020 insertions(+), 1067 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerRESTClient.java
----------------------------------------------------------------------
diff --git a/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerRESTClient.java b/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerRESTClient.java
index c311670..7cfd040 100644
--- a/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerRESTClient.java
+++ b/agents-common/src/main/java/org/apache/ranger/plugin/util/RangerRESTClient.java
@@ -212,8 +212,7 @@ public class RangerRESTClient {
client = Client.create(config);
}
- // TODO: for testing only
- if(!StringUtils.isEmpty(mUsername) || !StringUtils.isEmpty(mPassword)) {
+ if(!StringUtils.isEmpty(mUsername) && !StringUtils.isEmpty(mPassword)) {
client.addFilter(new HTTPBasicAuthFilter(mUsername, mPassword));
}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/agents-common/src/main/java/org/apache/ranger/plugin/util/ServiceTags.java
----------------------------------------------------------------------
diff --git a/agents-common/src/main/java/org/apache/ranger/plugin/util/ServiceTags.java b/agents-common/src/main/java/org/apache/ranger/plugin/util/ServiceTags.java
index 7800e9a..c3fb4cf 100644
--- a/agents-common/src/main/java/org/apache/ranger/plugin/util/ServiceTags.java
+++ b/agents-common/src/main/java/org/apache/ranger/plugin/util/ServiceTags.java
@@ -23,6 +23,8 @@ package org.apache.ranger.plugin.util;
import java.util.Date;
import java.util.List;
import java.util.Map;
+import java.util.HashMap;
+import java.util.ArrayList;
import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
@@ -51,7 +53,6 @@ public class ServiceTags implements java.io.Serializable {
public static final String TAGMODEL_SHARED = "shared";
public static final String TAGMODEL_RESOURCE_PRIVATE = "resource_private";
-
private String op = OP_ADD_OR_UPDATE;
private String tagModel = TAGMODEL_SHARED;
private String serviceName;
@@ -62,6 +63,22 @@ public class ServiceTags implements java.io.Serializable {
private List<RangerServiceResource> serviceResources;
private Map<Long, List<Long>> resourceToTagIds;
+ public ServiceTags() {
+ this(OP_ADD_OR_UPDATE, TAGMODEL_SHARED, null, 0L, null, null, null, null, null);
+ }
+
+ public ServiceTags(String op, String tagModel, String serviceName, Long tagVersion, Date tagUpdateTime, Map<Long, RangerTagDef> tagDefinitions,
+ Map<Long, RangerTag> tags, List<RangerServiceResource> serviceResources, Map<Long, List<Long>> resourceToTagIds) {
+ setOp(op);
+ setTagModel(tagModel);
+ setServiceName(serviceName);
+ setTagVersion(tagVersion);
+ setTagUpdateTime(tagUpdateTime);
+ setTagDefinitions(tagDefinitions);
+ setTags(tags);
+ setServiceResources(serviceResources);
+ setResourceToTagIds(resourceToTagIds);
+ }
/**
* @return the op
*/
@@ -138,7 +155,7 @@ public class ServiceTags implements java.io.Serializable {
}
public void setTagDefinitions(Map<Long, RangerTagDef> tagDefinitions) {
- this.tagDefinitions = tagDefinitions;
+ this.tagDefinitions = tagDefinitions == null ? new HashMap<Long, RangerTagDef>() : tagDefinitions;
}
public Map<Long, RangerTag> getTags() {
@@ -146,7 +163,7 @@ public class ServiceTags implements java.io.Serializable {
}
public void setTags(Map<Long, RangerTag> tags) {
- this.tags = tags;
+ this.tags = tags == null ? new HashMap<Long, RangerTag>() : tags;
}
public List<RangerServiceResource> getServiceResources() {
@@ -154,7 +171,7 @@ public class ServiceTags implements java.io.Serializable {
}
public void setServiceResources(List<RangerServiceResource> serviceResources) {
- this.serviceResources = serviceResources;
+ this.serviceResources = serviceResources == null ? new ArrayList<RangerServiceResource>() : serviceResources;
}
public Map<Long, List<Long>> getResourceToTagIds() {
@@ -162,7 +179,7 @@ public class ServiceTags implements java.io.Serializable {
}
public void setResourceToTagIds(Map<Long, List<Long>> resourceToTagIds) {
- this.resourceToTagIds = resourceToTagIds;
+ this.resourceToTagIds = resourceToTagIds == null ? new HashMap<Long, List<Long>>() : resourceToTagIds;
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/conf/templates/installprop2xml.properties
----------------------------------------------------------------------
diff --git a/tagsync/conf/templates/installprop2xml.properties b/tagsync/conf/templates/installprop2xml.properties
index 101a1ba..a6840b0 100644
--- a/tagsync/conf/templates/installprop2xml.properties
+++ b/tagsync/conf/templates/installprop2xml.properties
@@ -29,13 +29,15 @@ TAGADMIN_SSL_CONFIG_FILENAME = ranger.tagsync.tagadmin.rest.ssl.config.file
TAGSYNC_KEYSTORE_FILENAME = ranger.tagsync.tagadmin.keystore
-
+TAGSYNC_FILESOURCE_FILENAME = ranger.tagsync.filesource.filename
TAG_SOURCE_FILE_MOD_TIME_CHECK_INTERVAL = ranger.tagsync.filesource.modtime.check.interval
-TAGSYNC_FILESOURCE_FILENAME = ranger.tagsync.filesource.filename
+TAG_SOURCE_ATLAS_REST_URL = ranger.tagsync.atlasrestsource.endpoint
+TAG_SOURCE_ATLAS_REST_DOWNLOAD_INTERVAL = ranger.tagsync.atlasrestsource.download.interval
TAGSYNC_ATLAS_KAFKA_ENDPOINTS = atlas.kafka.bootstrap.servers
TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT = atlas.kafka.zookeeper.connect
TAGSYNC_ATLAS_CONSUMER_GROUP = atlas.kafka.entities.group.id
TAGSYNC_ATLAS_TO_RANGER_SERVICE_MAPPING = ranger.tagsync.atlas.to.service.mapping
+TAGSYNC_SOURCE_ATLAS_CUSTOM_RESOURCE_MAPPERS = ranger.tagsync.source.atlas.custom.resource.mappers
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/conf/templates/ranger-tagsync-template.xml
----------------------------------------------------------------------
diff --git a/tagsync/conf/templates/ranger-tagsync-template.xml b/tagsync/conf/templates/ranger-tagsync-template.xml
index 9a88681..bad71bd 100644
--- a/tagsync/conf/templates/ranger-tagsync-template.xml
+++ b/tagsync/conf/templates/ranger-tagsync-template.xml
@@ -32,6 +32,10 @@
<value></value>
</property>
<property>
+ <name>ranger.tagsync.atlasrestsource.download.interval</name>
+ <value></value>
+ </property>
+ <property>
<name>ranger.tagsync.tagadmin.rest.ssl.config.file</name>
<value></value>
</property>
@@ -59,4 +63,11 @@
<name>ranger.tagsync.atlas.to.service.mapping</name>
<value></value>
</property>
-</configuration>
+ <property>
+ <name>ranger.tagsync.atlasrestsource.endpoint</name>
+ <value></value>
+ </property>
+ <property>
+ <name>ranger.tagsync.source.atlas.custom.resource.mappers</name>
+ <value></value>
+ </property></configuration>
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/scripts/install.properties
----------------------------------------------------------------------
diff --git a/tagsync/scripts/install.properties b/tagsync/scripts/install.properties
index 95e87e5..b6665d1 100644
--- a/tagsync/scripts/install.properties
+++ b/tagsync/scripts/install.properties
@@ -37,9 +37,13 @@ TAGADMIN_ENDPOINT = http://localhost:6080
# SSL config file name for TagAdmin
TAGADMIN_SSL_CONFIG_FILENAME =
-# Source for tags (either 'atlas' or 'file')
+# Source for tags (either 'atlas' or 'file' or 'atlasrest')
TAG_SOURCE = atlas
+TAG_SOURCE_ATLAS_REST_URL = http://localhost:21000
+
+# Interval for checking the source for any changes in case of TAG_SOURCE = atlasrest
+TAG_SOURCE_ATLAS_REST_DOWNLOAD_INTERVAL = 900000
# File name to be used for reading tags information if TAG_SOURCE = file
@@ -73,3 +77,9 @@ TAGSYNC_ATLAS_CONSUMER_GROUP = ranger_entities_consumer
#
TAGSYNC_ATLAS_TO_RANGER_SERVICE_MAPPING=
+
+# A comma separated list of custom mapper class names which convert Atlas entities to
+# RangerServiceResource structures are specified here. If there are no custom mappers,
+# then it can be left blank
+
+TAGSYNC_SOURCE_ATLAS_CUSTOM_RESOURCE_MAPPERS=
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/scripts/ranger-tagsync-services.sh
----------------------------------------------------------------------
diff --git a/tagsync/scripts/ranger-tagsync-services.sh b/tagsync/scripts/ranger-tagsync-services.sh
index ca82ead..058ebf3 100755
--- a/tagsync/scripts/ranger-tagsync-services.sh
+++ b/tagsync/scripts/ranger-tagsync-services.sh
@@ -16,8 +16,8 @@
# limitations under the License.
if [[ -z $1 ]]; then
- echo "Invalid argument [$1];"
- echo "Usage: Only start | stop | restart | version, are supported."
+ echo "No argument provided.."
+ echo "Usage: $0 {start | stop | restart | version}"
exit;
fi
action=$1
@@ -119,7 +119,7 @@ elif [ "${action}" == "VERSION" ]; then
exit
else
echo "Invalid argument [$1];"
- echo "Usage: Only start | stop | restart | version, are supported."
+ echo "Usage: $0 {start | stop | restart | version}"
exit;
fi
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/scripts/ranger-tagsync-sync.sh
----------------------------------------------------------------------
diff --git a/tagsync/scripts/ranger-tagsync-sync.sh b/tagsync/scripts/ranger-tagsync-sync.sh
new file mode 100755
index 0000000..f8990e3
--- /dev/null
+++ b/tagsync/scripts/ranger-tagsync-sync.sh
@@ -0,0 +1,66 @@
+#!/bin/bash
+
+# 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.
+
+if [[ -z $1 ]]; then
+ echo "No argument provided.."
+ echo "Usage: $0 {FILE | ATLAS_DOWNLOAD}"
+ exit -1;
+fi
+action=$1
+action=`echo $action | tr '[:lower:]' '[:upper:]'`
+realScriptPath=`readlink -f $0`
+realScriptDir=`dirname $realScriptPath`
+cd $realScriptDir
+cdir=`pwd`
+
+if [ "${action}" == "FILE" ]; then
+ action=file
+elif [ "${action}" == "ATLAS_DOWNLOAD" ]; then
+ action=atlasrest
+else
+ echo "Invalid argument [$action];"
+ echo "Usage: $0 {FILE | ATLAS_DOWNLOAD}"
+ exit -1;
+fi
+
+if [ -f ${cdir}/conf/java_home.sh ]; then
+ . ${cdir}/conf/java_home.sh
+fi
+
+for custom_env_script in `find ${cdir}/conf.dist/ -name "ranger-tagsync-env*"`; do
+ if [ -f $custom_env_script ]; then
+ . $custom_env_script
+ fi
+done
+
+if [ "$JAVA_HOME" != "" ]; then
+ export PATH=$JAVA_HOME/bin:$PATH
+fi
+
+logdir=/var/log/ranger/tagsync-$action
+
+if [ ! -d $logdir ]; then
+ mkdir -p $logdir
+ chmod 777 $logdir
+fi
+
+cp="${cdir}/conf:${cdir}/dist/*:${cdir}/lib/*"
+
+cd ${cdir}
+umask 0077
+java -Dproc_rangertagsync-$action ${JAVA_OPTS} -Dlogdir="${logdir}" -cp "${cp}" org.apache.ranger.tagsync.process.TagSynchronizer $action > ${logdir}/tagsync.out 2>&1
+
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/scripts/setup.py
----------------------------------------------------------------------
diff --git a/tagsync/scripts/setup.py b/tagsync/scripts/setup.py
index 1c8552b..c100f73 100755
--- a/tagsync/scripts/setup.py
+++ b/tagsync/scripts/setup.py
@@ -77,8 +77,8 @@ TAGSYNC_INSTALL_PROP_PREFIX_FOR_ATLAS_RANGER_MAPPING = 'ranger.tagsync.atlas.'
TAGSYNC_ATLAS_CLUSTER_IDENTIFIER = '.instance.'
TAGSYNC_INSTALL_PROP_SUFFIX_FOR_ATLAS_RANGER_MAPPING = '.ranger.service'
TAG_SOURCE_ATLAS = 'atlas'
+TAG_SOURCE_ATLASREST = 'atlasrest'
TAG_SOURCE_FILE = 'file'
-TAG_SOURCE_LIST = [ TAG_SOURCE_ATLAS, TAG_SOURCE_FILE ]
def archiveFile(originalFileName):
archiveDir = dirname(originalFileName)
@@ -199,20 +199,11 @@ def convertInstallPropsToXML(props):
else:
print "Direct Key not found:%s" % (k)
- ret['ranger.tagsync.sink.impl.class'] = 'org.apache.ranger.sink.policymgr.TagRESTSink'
+ ret['ranger.tagsync.sink.impl.class'] = 'org.apache.ranger.sink.policymgr.TagAdminRESTSink'
+
if (TAG_SOURCE_KEY in ret):
- tagSource = ret[TAG_SOURCE_KEY]
- if (tagSource == TAG_SOURCE_ATLAS):
- ret['ranger.tagsync.source.impl.class'] = 'atlas'
- elif (tagSource == TAG_SOURCE_FILE):
- ret['ranger.tagsync.source.impl.class'] = 'file'
- else:
- print "ERROR: Invalid value (%s) defined for %s in install.properties. Only valid values are %s" % (tagSource, TAG_SOURCE_ATLAS,TAG_SOURCE_FILE)
- sys.exit(1)
- del ret['TAG_SOURCE']
- else:
- print "ERROR: No value defined for TAG_SOURCE in install.properties. valid values are %s" % (TAG_SOURCE_ATLAS, TAG_SOURCE_FILE)
- sys.exit(1)
+ ret['ranger.tagsync.source.impl.class'] = ret[TAG_SOURCE_KEY]
+ del ret[TAG_SOURCE_KEY]
atlasOutFile.close()
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/model/AbstractTagSource.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/model/AbstractTagSource.java b/tagsync/src/main/java/org/apache/ranger/tagsync/model/AbstractTagSource.java
new file mode 100644
index 0000000..d6baeb2
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/model/AbstractTagSource.java
@@ -0,0 +1,65 @@
+/*
+ * 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.ranger.tagsync.model;
+
+import com.google.gson.Gson;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.ranger.plugin.util.ServiceTags;
+
+public abstract class AbstractTagSource implements TagSource {
+ private static final Log LOG = LogFactory.getLog(AbstractTagSource.class);
+ private TagSink tagSink;
+ protected boolean shutdown = false;
+
+ @Override
+ public void setTagSink(TagSink sink) {
+ if (sink == null) {
+ LOG.error("Sink is null!!!");
+ } else {
+ this.tagSink = sink;
+ }
+ }
+ @Override
+ public void synchUp() {}
+
+ public void updateSink(final ServiceTags serviceTags) {
+ if (serviceTags == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("No ServiceTags to upload");
+ }
+ } else {
+ if (LOG.isDebugEnabled()) {
+ String serviceTagsJSON = new Gson().toJson(serviceTags);
+ LOG.debug("Uploading serviceTags=" + serviceTagsJSON);
+ }
+
+ try {
+ tagSink.uploadServiceTags(serviceTags);
+ } catch (Exception exception) {
+ LOG.error("uploadServiceTags() failed..", exception);
+ }
+ }
+ }
+
+ public void stop() {
+ shutdown = true;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/model/TagSource.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/model/TagSource.java b/tagsync/src/main/java/org/apache/ranger/tagsync/model/TagSource.java
index 2df8036..7d19562 100644
--- a/tagsync/src/main/java/org/apache/ranger/tagsync/model/TagSource.java
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/model/TagSource.java
@@ -27,9 +27,11 @@ public interface TagSource {
void setTagSink(TagSink sink);
- void updateSink() throws Exception;
+ void synchUp();
- Thread start();
+ boolean start();
+
+ void stop();
boolean isChanged();
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSyncConfig.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSyncConfig.java b/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSyncConfig.java
index e1b5130..0dfea25 100644
--- a/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSyncConfig.java
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSyncConfig.java
@@ -47,22 +47,29 @@ public class TagSyncConfig extends Configuration {
private static final String TAGSYNC_FILESOURCE_MOD_TIME_CHECK_INTERVAL_PROP = "ranger.tagsync.filesource.modtime.check.interval";
+ private static final String TAGSYNC_ATLAS_REST_SOURCE_DOWNLOAD_INTERVAL_PROP = "ranger.tagsync.atlasrestsource.download.interval";
+
private static final String TAGSYNC_SOURCE_CLASS_PROP = "ranger.tagsync.source.impl.class";
private static final String TAGSYNC_SINK_CLASS_PROP = "ranger.tagsync.sink.impl.class";
- private static final String TAGSYNC_ATLASSOURCE_ENDPOINT_PROP = "ranger.tagsync.atlassource.endpoint";
+ private static final String TAGSYNC_ATLASSOURCE_ENDPOINT_PROP = "ranger.tagsync.atlasrestsource.endpoint";
- private static final String TAGSYNC_SERVICENAME_MAPPER_PROP_PREFIX = "ranger.tagsync.atlas.";
+ public static final String TAGSYNC_SERVICENAME_MAPPER_PROP_PREFIX = "ranger.tagsync.atlas.";
- private static final String TAGSYNC_SERVICENAME_MAPPER_PROP_SUFFIX = ".ranger.service";
+ public static final String TAGSYNC_SERVICENAME_MAPPER_PROP_SUFFIX = ".ranger.service";
- private static final String TAGSYNC_DEFAULT_CLUSTERNAME_AND_COMPONENTNAME_SEPARATOR = "_";
+ public static final String TAGSYNC_DEFAULT_CLUSTERNAME_AND_COMPONENTNAME_SEPARATOR = "_";
private static final String TAGSYNC_TAGADMIN_KEYSTORE_PROP = "ranger.tagsync.tagadmin.keystore";
private static final String TAGSYNC_TAGADMIN_ALIAS_PROP = "ranger.tagsync.tagadmin.alias";
private static final String TAGSYNC_TAGADMIN_PASSWORD_PROP = "ranger.tagsync.tagadmin.password";
private static final String DEFAULT_TAGADMIN_USERNAME = "rangertagsync";
+ private static final String TAGSYNC_SOURCE_ATLAS_CUSTOM_RESOURCE_MAPPERS_PROP = "ranger.tagsync.source.atlas.custom.resource.mappers";
+
+ private static final long DEFAULT_TAGSYNC_FILESOURCE_MOD_TIME_CHECK_INTERVAL = 60000;
+
+ private static final long DEFAULT_TAGSYNC_REST_SOURCE_DOWNLOAD_INTERVAL = 900000;
public static TagSyncConfig getInstance() {
TagSyncConfig newConfig = new TagSyncConfig();
@@ -158,7 +165,7 @@ public class TagSyncConfig extends Configuration {
static public boolean isTagSyncEnabled(Properties prop) {
String val = prop.getProperty(TAGSYNC_ENABLED_PROP);
- return !(val != null && val.trim().equalsIgnoreCase("falae"));
+ return !(val != null && val.trim().equalsIgnoreCase("false"));
}
static public String getTagSyncLogdir(Properties prop) {
@@ -168,17 +175,39 @@ public class TagSyncConfig extends Configuration {
static public long getTagSourceFileModTimeCheckIntervalInMillis(Properties prop) {
String val = prop.getProperty(TAGSYNC_FILESOURCE_MOD_TIME_CHECK_INTERVAL_PROP);
- return Long.valueOf(val);
+ long ret = DEFAULT_TAGSYNC_FILESOURCE_MOD_TIME_CHECK_INTERVAL;
+ if (StringUtils.isNotBlank(val)) {
+ try {
+ ret = Long.valueOf(val);
+ } catch (NumberFormatException exception) {
+ // Ignore
+ }
+ }
+ return ret;
+ }
+
+ static public long getTagSourceAtlasDownloadIntervalInMillis(Properties prop) {
+ String val = prop.getProperty(TAGSYNC_ATLAS_REST_SOURCE_DOWNLOAD_INTERVAL_PROP);
+ long ret = DEFAULT_TAGSYNC_REST_SOURCE_DOWNLOAD_INTERVAL;
+ if (StringUtils.isNotBlank(val)) {
+ try {
+ ret = Long.valueOf(val);
+ } catch (NumberFormatException exception) {
+ // Ignore
+ }
+ }
+ return ret;
}
- static public String getTagSourceClassName(Properties prop) {
- String val = prop.getProperty(TAGSYNC_SOURCE_CLASS_PROP);
- if (StringUtils.equalsIgnoreCase(val, "atlas")) {
- return "org.apache.ranger.tagsync.source.atlas.TagAtlasSource";
- } else if (StringUtils.equalsIgnoreCase(val, "file")) {
- return "org.apache.ranger.tagsync.source.file.TagFileSource";
+ static public String getTagSourceClassName(String sourceName) {
+ if (StringUtils.equalsIgnoreCase(sourceName, "atlas")) {
+ return "org.apache.ranger.tagsync.source.atlas.AtlasTagSource";
+ } else if (StringUtils.equalsIgnoreCase(sourceName, "file")) {
+ return "org.apache.ranger.tagsync.source.file.FileTagSource";
+ } else if (StringUtils.equalsIgnoreCase(sourceName, "atlasrest")) {
+ return "org.apache.ranger.tagsync.source.atlasrest.AtlasRESTTagSource";
} else
- return val;
+ return sourceName;
}
static public String getTagSource(Properties prop) {
@@ -188,7 +217,7 @@ public class TagSyncConfig extends Configuration {
static public String getTagSinkClassName(Properties prop) {
String val = prop.getProperty(TAGSYNC_SINK_CLASS_PROP);
if (StringUtils.equalsIgnoreCase(val, "tagadmin")) {
- return "org.apache.ranger.tagsync.sink.tagadmin.TagRESTSink";
+ return "org.apache.ranger.tagsync.sink.tagadmin.TagAdminRESTSink";
} else
return val;
}
@@ -250,15 +279,8 @@ public class TagSyncConfig extends Configuration {
return "";
}
- static public String getServiceName(String componentName, String instanceName, Properties prop) {
- String propName = TAGSYNC_SERVICENAME_MAPPER_PROP_PREFIX + componentName
- + ".instance." + instanceName
- + TAGSYNC_SERVICENAME_MAPPER_PROP_SUFFIX;
- String val = prop.getProperty(propName);
- if (StringUtils.isBlank(val)) {
- val = instanceName + TAGSYNC_DEFAULT_CLUSTERNAME_AND_COMPONENTNAME_SEPARATOR + componentName;
- }
- return val;
+ static public String getCustomAtlasResourceMappers(Properties prop) {
+ return prop.getProperty(TAGSYNC_SOURCE_ATLAS_CUSTOM_RESOURCE_MAPPERS_PROP);
}
private TagSyncConfig() {
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java b/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java
index 7bae973..c1a1c39 100644
--- a/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/process/TagSynchronizer.java
@@ -19,49 +19,50 @@
package org.apache.ranger.tagsync.process;
+import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.apache.ranger.tagsync.model.TagSink;
import org.apache.ranger.tagsync.model.TagSource;
-import java.util.Map;
-import java.util.Properties;
+import java.util.*;
public class TagSynchronizer {
private static final Logger LOG = Logger.getLogger(TagSynchronizer.class);
private boolean shutdownFlag = false;
- private TagSource tagSource = null;
+ private List<TagSource> tagSources;
private Properties properties = null;
public static void main(String[] args) {
- boolean tagSynchronizerInitialized = false;
TagSynchronizer tagSynchronizer = new TagSynchronizer();
- while (!tagSynchronizerInitialized) {
+ TagSyncConfig config = TagSyncConfig.getInstance();
+ Properties props = config.getProperties();
- TagSyncConfig config = TagSyncConfig.getInstance();
- Properties props = config.getProperties();
+ tagSynchronizer.setProperties(props);
- tagSynchronizer.setProperties(props);
- tagSynchronizerInitialized = tagSynchronizer.initialize();
+ String tagSourceName = null;
+ if (args.length > 0) {
+ tagSourceName = args[0];
+ LOG.info("TagSource is set to " + args[0]);
+ }
- if (!tagSynchronizerInitialized) {
- LOG.error("TagSynchronizer failed to initialize correctly");
+ boolean tagSynchronizerInitialized = tagSynchronizer.initialize(tagSourceName);
- try {
- LOG.error("Sleeping for [60] seconds before attempting to re-read configuration XML files");
- Thread.sleep(60 * 1000);
- } catch (InterruptedException e) {
- LOG.error("Failed to wait for [60] seconds", e);
- }
+ if (tagSynchronizerInitialized) {
+ if (StringUtils.isNotBlank(tagSourceName)) {
+ tagSynchronizer.getFirstTagSource().synchUp();
+ } else {
+ tagSynchronizer.run();
}
+ } else {
+ LOG.error("TagSynchronizer failed to initialize correctly, exiting..");
+ System.exit(-1);
}
-
- tagSynchronizer.run();
}
public TagSynchronizer() {
@@ -80,64 +81,124 @@ public class TagSynchronizer {
}
}
- public boolean initialize() {
+ public boolean initialize(String source) {
if (LOG.isDebugEnabled()) {
- LOG.debug("==> TagSynchronizer.initialize()");
+ LOG.debug("==> TagSynchronizer.initialize(" + source + ")");
}
printConfigurationProperties();
- boolean ret = true;
+ boolean ret = false;
+
+ String tagSourceNames = StringUtils.isNotBlank(source) ? source :
+ TagSyncConfig.getTagSource(properties);
+
+ if (StringUtils.isNotBlank(tagSourceNames)) {
+
+ LOG.info("Initializing TAG source and sink");
- String tagSourceName = TagSyncConfig.getTagSource(properties);
+ TagSink tagSink = initializeTagSink();
- if (StringUtils.isBlank(tagSourceName) ||
- (!StringUtils.equalsIgnoreCase(tagSourceName, "file") && !StringUtils.equalsIgnoreCase(tagSourceName, "atlas"))) {
- ret = false;
- LOG.error("'ranger.tagsync.source.impl.class' value is invalid!, 'ranger.tagsync.source.impl.class'=" + tagSourceName + ". Supported 'ranger.tagsync.source.impl.class' values are : file, atlas");
+ if (tagSink != null) {
+
+ tagSources = initializeTagSources(tagSourceNames);
+
+ if (CollectionUtils.isNotEmpty(tagSources)) {
+ for (TagSource tagSource : tagSources) {
+ tagSource.setTagSink(tagSink);
+ }
+ ret = true;
+ }
+ }
+ } else {
+ LOG.error("'ranger.tagsync.source.impl.class' value is not specified or is empty!");
}
- if (ret) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== TagSynchronizer.initialize(" + tagSourceNames + ") : " + ret);
+ }
+
+ return ret;
+ }
+
+ public TagSink initializeTagSink() {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> TagSynchronizer.initializeTagSink()");
+ }
- try {
- LOG.info("Initializing TAG source and sink");
- // Initialize tagSink and tagSource
- String tagSourceClassName = TagSyncConfig.getTagSourceClassName(properties);
- String tagSinkClassName = TagSyncConfig.getTagSinkClassName(properties);
+ TagSink ret = null;
+ try {
+ String tagSinkClassName = TagSyncConfig.getTagSinkClassName(properties);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("tagSinkClassName=" + tagSinkClassName);
+ }
+ @SuppressWarnings("unchecked")
+ Class<TagSink> tagSinkClass = (Class<TagSink>) Class.forName(tagSinkClassName);
+
+ ret = tagSinkClass.newInstance();
+
+ if (!ret.initialize(properties)) {
+ LOG.error("Failed to initialize TAG sink " + tagSinkClassName);
+
+ ret = null;
+ }
+ } catch (Throwable t) {
+ LOG.error("Failed to initialize TAG sink. Error details: ", t);
+ ret = null;
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== TagSynchronizer.initializeTagSink()");
+ }
+ return ret;
+ }
+
+ public List<TagSource> initializeTagSources(String tagSourceNames) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> TagSynchronizer.initializeTagSources(" + tagSourceNames + ")");
+ }
+
+ List<TagSource> ret = new ArrayList<TagSource>();
+
+ String[] tagSourcesArray = tagSourceNames.split(",");
+
+ List<String> tagSourceList = Arrays.asList(tagSourcesArray);
+
+ try {
+ for (String tagSourceName : tagSourceList) {
+
+ String tagSourceClassName = TagSyncConfig.getTagSourceClassName(tagSourceName.trim());
if (LOG.isDebugEnabled()) {
- LOG.debug("tagSourceClassName=" + tagSourceClassName + ", tagSinkClassName=" + tagSinkClassName);
+ LOG.debug("tagSourceClassName=" + tagSourceClassName);
}
@SuppressWarnings("unchecked")
Class<TagSource> tagSourceClass = (Class<TagSource>) Class.forName(tagSourceClassName);
-
- @SuppressWarnings("unchecked")
- Class<TagSink> tagSinkClass = (Class<TagSink>) Class.forName(tagSinkClassName);
-
- TagSink tagSink = tagSinkClass.newInstance();
- tagSource = tagSourceClass.newInstance();
+ TagSource tagSource = tagSourceClass.newInstance();
if (LOG.isDebugEnabled()) {
- LOG.debug("Created instance of " + tagSourceClassName + ", " + tagSinkClassName);
+ LOG.debug("Created instance of " + tagSourceClassName);
}
- ret = tagSink.initialize(properties) && tagSource.initialize(properties);
+ if (!tagSource.initialize(properties)) {
+ LOG.error("Failed to initialize TAG source " + tagSourceClassName);
- if (ret) {
- tagSource.setTagSink(tagSink);
+ ret.clear();
+ break;
}
-
- LOG.info("Done initializing TAG source and sink");
- } catch (Throwable t) {
- LOG.error("Failed to initialize TAG source/sink. Error details: ", t);
- ret = false;
+ ret.add(tagSource);
}
+
+ } catch (Throwable t) {
+ LOG.error("Failed to initialize TAG sources. Error details: ", t);
+ ret.clear();
}
if (LOG.isDebugEnabled()) {
- LOG.debug("<== TagSynchronizer.initialize(), result=" + ret);
+ LOG.debug("<== TagSynchronizer.initializeTagSources(" + tagSourceNames + ")");
}
return ret;
@@ -150,35 +211,31 @@ public class TagSynchronizer {
long shutdownCheckIntervalInMs = 60*1000;
- Thread tagSourceThread = null;
+ boolean tagSourcesStarted = true;
try {
- tagSourceThread = tagSource.start();
+ for (TagSource tagSource : tagSources) {
+ tagSourcesStarted = tagSourcesStarted && tagSource.start();
+ }
- if (tagSourceThread != null) {
+ if (tagSourcesStarted) {
while (!shutdownFlag) {
try {
LOG.debug("Sleeping for [" + shutdownCheckIntervalInMs + "] milliSeconds");
Thread.sleep(shutdownCheckIntervalInMs);
} catch (InterruptedException e) {
- LOG.error("Failed to wait for [" + shutdownCheckIntervalInMs + "] milliseconds before attempting to synchronize tag information", e);
+ LOG.error("Failed to wait for [" + shutdownCheckIntervalInMs + "] milliseconds before attempting to synchronize tag information ", e);
+ break;
}
}
}
} catch (Throwable t) {
- LOG.error("tag-sync thread got an error", t);
+ LOG.error("tag-sync main thread got an error", t);
} finally {
- if (tagSourceThread != null) {
- LOG.error("Shutting down the tag-sync thread");
- LOG.info("Interrupting tagSourceThread...");
- tagSourceThread.interrupt();
- try {
- tagSourceThread.join();
- } catch (InterruptedException interruptedException) {
- LOG.error("tagSourceThread.join() was interrupted");
- }
- } else {
- LOG.error("Could not start tagSource monitoring thread");
+ LOG.info("Stopping all tagSources");
+
+ for (TagSource tagSource : tagSources) {
+ tagSource.stop();
}
}
@@ -192,6 +249,10 @@ public class TagSynchronizer {
this.shutdownFlag = true;
}
+ public TagSource getFirstTagSource() {
+ return tagSources.get(0);
+ }
+
public void printConfigurationProperties() {
LOG.info("--------------------------------");
LOG.info("");
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagAdminRESTSink.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagAdminRESTSink.java b/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagAdminRESTSink.java
new file mode 100644
index 0000000..1541034
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagAdminRESTSink.java
@@ -0,0 +1,162 @@
+/*
+ * 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.ranger.tagsync.sink.tagadmin;
+
+import com.sun.jersey.api.client.ClientResponse;
+import com.sun.jersey.api.client.WebResource;
+
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.ranger.admin.client.datatype.RESTResponse;
+import org.apache.ranger.tagsync.model.TagSink;
+import org.apache.ranger.plugin.util.RangerRESTClient;
+import org.apache.ranger.plugin.util.SearchFilter;
+import org.apache.ranger.plugin.util.ServiceTags;
+import org.apache.ranger.tagsync.process.TagSyncConfig;
+
+import java.util.Map;
+import java.util.Properties;
+
+public class TagAdminRESTSink implements TagSink {
+ private static final Log LOG = LogFactory.getLog(TagAdminRESTSink.class);
+
+ private static final String REST_PREFIX = "/service";
+ private static final String MODULE_PREFIX = "/tags";
+
+ private static final String REST_MIME_TYPE_JSON = "application/json" ;
+
+ private static final String REST_URL_IMPORT_SERVICETAGS_RESOURCE = REST_PREFIX + MODULE_PREFIX + "/importservicetags/";
+
+ private RangerRESTClient tagRESTClient = null;
+
+ @Override
+ public boolean initialize(Properties properties) {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("==> TagAdminRESTSink.initialize()");
+ }
+
+ boolean ret = true;
+
+ String restUrl = TagSyncConfig.getTagAdminRESTUrl(properties);
+ String sslConfigFile = TagSyncConfig.getTagAdminRESTSslConfigFile(properties);
+ String userName = TagSyncConfig.getTagAdminUserName(properties);
+ String password = TagSyncConfig.getTagAdminPassword(properties);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("restUrl=" + restUrl);
+ LOG.debug("sslConfigFile=" + sslConfigFile);
+ LOG.debug("userName=" + userName);
+ }
+
+ if (StringUtils.isBlank(restUrl)) {
+ ret = false;
+ LOG.error("No value specified for property 'ranger.tagsync.tagadmin.rest.url'!");
+ } else {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("ranger.tagsync.tagadmin.rest.url:" + restUrl);
+ }
+ }
+
+ if (ret) {
+ tagRESTClient = new RangerRESTClient(restUrl, sslConfigFile);
+ tagRESTClient.setBasicAuthInfo(userName, password);
+
+ ret = testConnection();
+ }
+
+ if (!ret) {
+ LOG.error("Cannot connect to Tag Admin. Please recheck configuration properties and/or check if Tag Admin is running and responsive");
+ }
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("<== TagAdminRESTSink.initialize(), result=" + ret);
+ }
+
+ return ret;
+ }
+
+ public boolean testConnection() {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> TagAdminRESTSink.testConnection()");
+ }
+
+ boolean ret = true;
+
+ try {
+ // build a dummy serviceTags structure and upload it to test connectivity
+ ServiceTags serviceTags = new ServiceTags();
+ serviceTags.setOp(ServiceTags.OP_ADD_OR_UPDATE);
+ uploadServiceTags(serviceTags);
+ } catch (Exception exception) {
+ LOG.error("test-upload of serviceTags failed.", exception);
+ ret = false;
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== TagAdminRESTSink.testConnection(), result=" + ret);
+ }
+ return ret;
+ }
+
+ @Override
+ synchronized public void uploadServiceTags(ServiceTags serviceTags) throws Exception {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("==> uploadServiceTags()");
+ }
+ WebResource webResource = createWebResource(REST_URL_IMPORT_SERVICETAGS_RESOURCE);
+
+ ClientResponse response = webResource.accept(REST_MIME_TYPE_JSON).type(REST_MIME_TYPE_JSON).put(ClientResponse.class, tagRESTClient.toJson(serviceTags));
+
+ if(response == null || response.getStatus() != 204) {
+ LOG.error("RangerAdmin REST call returned with response={" + response + "}");
+
+ RESTResponse resp = RESTResponse.fromClientResponse(response);
+
+ LOG.error("Upload of service-tags failed with message " + resp.getMessage());
+
+ throw new Exception("Upload of service-tags failed with response: " + response);
+ }
+
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("<== uploadServiceTags()");
+ }
+ }
+
+ private WebResource createWebResource(String url) {
+ return createWebResource(url, null);
+ }
+
+ private WebResource createWebResource(String url, SearchFilter filter) {
+ WebResource ret = tagRESTClient.getResource(url);
+
+ if(filter != null && !MapUtils.isEmpty(filter.getParams())) {
+ for(Map.Entry<String, String> e : filter.getParams().entrySet()) {
+ String name = e.getKey();
+ String value = e.getValue();
+
+ ret.queryParam(name, value);
+ }
+ }
+
+ return ret;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagRESTSink.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagRESTSink.java b/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagRESTSink.java
deleted file mode 100644
index 41085d0..0000000
--- a/tagsync/src/main/java/org/apache/ranger/tagsync/sink/tagadmin/TagRESTSink.java
+++ /dev/null
@@ -1,160 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-package org.apache.ranger.tagsync.sink.tagadmin;
-
-import com.sun.jersey.api.client.ClientResponse;
-import com.sun.jersey.api.client.WebResource;
-
-import org.apache.commons.collections.MapUtils;
-import org.apache.commons.lang.StringUtils;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.ranger.admin.client.datatype.RESTResponse;
-import org.apache.ranger.tagsync.model.TagSink;
-import org.apache.ranger.plugin.util.RangerRESTClient;
-import org.apache.ranger.plugin.util.SearchFilter;
-import org.apache.ranger.plugin.util.ServiceTags;
-import org.apache.ranger.tagsync.process.TagSyncConfig;
-
-import java.util.Map;
-import java.util.Properties;
-
-public class TagRESTSink implements TagSink {
- private static final Log LOG = LogFactory.getLog(TagRESTSink.class);
-
- private static final String REST_PREFIX = "/service";
- private static final String MODULE_PREFIX = "/tags";
-
- private static final String REST_MIME_TYPE_JSON = "application/json" ;
-
- private static final String REST_URL_IMPORT_SERVICETAGS_RESOURCE = REST_PREFIX + MODULE_PREFIX + "/importservicetags/";
-
- private RangerRESTClient tagRESTClient = null;
-
- @Override
- public boolean initialize(Properties properties) {
- if(LOG.isDebugEnabled()) {
- LOG.debug("==> TagRESTSink.initialize()");
- }
-
- boolean ret = true;
-
- String restUrl = TagSyncConfig.getTagAdminRESTUrl(properties);
- String sslConfigFile = TagSyncConfig.getTagAdminRESTSslConfigFile(properties);
- String userName = TagSyncConfig.getTagAdminUserName(properties);
- String password = TagSyncConfig.getTagAdminPassword(properties);
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("restUrl=" + restUrl);
- LOG.debug("sslConfigFile=" + sslConfigFile);
- LOG.debug("userName=" + userName);
- }
-
- if (StringUtils.isBlank(restUrl)) {
- ret = false;
- LOG.error("No value specified for property 'ranger.tagsync.tagadmin.rest.url'!");
- } else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("ranger.tagsync.tagadmin.rest.url:" + restUrl);
- }
- }
-
- if (ret) {
- tagRESTClient = new RangerRESTClient(restUrl, sslConfigFile);
- tagRESTClient.setBasicAuthInfo(userName, password);
-
- ret = testConnection();
- }
-
- if (!ret) {
- LOG.error("Cannot connect to Tag Admin. Please recheck configuration properties and/or check if Tag Admin is running and responsive");
- }
-
- if(LOG.isDebugEnabled()) {
- LOG.debug("<== TagRESTSink.initialize(), result=" + ret);
- }
-
- return ret;
- }
-
- public boolean testConnection() {
- if (LOG.isDebugEnabled()) {
- LOG.debug("==> TagRESTSink.testConnection()");
- }
-
- boolean ret = true;
-
- try {
- // build a dummy serviceTags structure and upload it to test connectivity
- ServiceTags serviceTags = new ServiceTags();
- serviceTags.setOp(ServiceTags.OP_ADD_OR_UPDATE);
- uploadServiceTags(serviceTags);
- } catch (Exception exception) {
- LOG.error("test-upload of serviceTags failed.", exception);
- ret = false;
- }
-
- if (LOG.isDebugEnabled()) {
- LOG.debug("<== TagRESTSink.testConnection(), result=" + ret);
- }
- return ret;
- }
-
- @Override
- public void uploadServiceTags(ServiceTags serviceTags) throws Exception {
- if(LOG.isDebugEnabled()) {
- LOG.debug("==> uploadServiceTags()");
- }
- WebResource webResource = createWebResource(REST_URL_IMPORT_SERVICETAGS_RESOURCE);
-
- ClientResponse response = webResource.accept(REST_MIME_TYPE_JSON).type(REST_MIME_TYPE_JSON).put(ClientResponse.class, tagRESTClient.toJson(serviceTags));
-
- if(response == null || response.getStatus() != 204) {
- LOG.error("RangerAdmin REST call returned with response={" + response + "}");
-
- RESTResponse resp = RESTResponse.fromClientResponse(response);
-
- LOG.error("Upload of service-tags failed with message " + resp.getMessage());
- }
-
- if(LOG.isDebugEnabled()) {
- LOG.debug("<== uploadServiceTags()");
- }
- }
-
- private WebResource createWebResource(String url) {
- return createWebResource(url, null);
- }
-
- private WebResource createWebResource(String url, SearchFilter filter) {
- WebResource ret = tagRESTClient.getResource(url);
-
- if(filter != null && !MapUtils.isEmpty(filter.getParams())) {
- for(Map.Entry<String, String> e : filter.getParams().entrySet()) {
- String name = e.getKey();
- String value = e.getValue();
-
- ret.queryParam(name, value);
- }
- }
-
- return ret;
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasHiveResourceMapper.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasHiveResourceMapper.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasHiveResourceMapper.java
new file mode 100644
index 0000000..a17d611
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasHiveResourceMapper.java
@@ -0,0 +1,203 @@
+/*
+ * 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.ranger.tagsync.source.atlas;
+
+import org.apache.atlas.typesystem.IReferenceableInstance;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.ranger.plugin.model.RangerPolicy;
+import org.apache.ranger.plugin.model.RangerServiceResource;
+import org.apache.ranger.tagsync.process.TagSyncConfig;
+
+import java.util.Map;
+import java.util.HashMap;
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Arrays;
+
+public class AtlasHiveResourceMapper extends AtlasResourceMapper {
+ private static final Log LOG = LogFactory.getLog(AtlasHiveResourceMapper.class);
+
+ public static final String COMPONENT_NAME = "hive";
+
+ public static final String ENTITY_TYPE_HIVE_DB = "hive_db";
+ public static final String ENTITY_TYPE_HIVE_TABLE = "hive_table";
+ public static final String ENTITY_TYPE_HIVE_COLUMN = "hive_column";
+
+ public static final String RANGER_TYPE_HIVE_DB = "database";
+ public static final String RANGER_TYPE_HIVE_TABLE = "table";
+ public static final String RANGER_TYPE_HIVE_COLUMN = "column";
+
+ public static final String ENTITY_ATTRIBUTE_QUALIFIED_NAME_FOR_HIVE_TABLE = "name";
+
+ static protected final String ENTITY_ATTRIBUTE_QUALIFIED_NAME = "qualifiedName";
+
+ private static String clusterDelimiter = "@";
+
+ private static String qualifiedNameDelimiter = "\\.";
+
+ public static final String[] supportedEntityTypes = { ENTITY_TYPE_HIVE_DB, ENTITY_TYPE_HIVE_TABLE, ENTITY_TYPE_HIVE_COLUMN };
+
+ public AtlasHiveResourceMapper() {
+ super();
+ }
+
+ @Override
+ public List<String> getSupportedEntityTypes() {
+ return Arrays.asList(supportedEntityTypes);
+ }
+
+ @Override
+ public RangerServiceResource buildResource(final IReferenceableInstance entity) throws Exception {
+
+ Map<String, RangerPolicy.RangerPolicyResource> elements = new HashMap<String, RangerPolicy.RangerPolicyResource>();
+
+ String serviceName = null;
+
+ List<String> components = getQualifiedNameComponents(entity);
+ // components should contain qualifiedName, clusterName, dbName, tableName, columnName in that order
+
+ String entityTypeName = entity.getTypeName();
+
+ String qualifiedName = components.get(0);
+
+ String clusterName, dbName, tableName, columnName;
+
+ if (components.size() > 1) {
+ clusterName = components.get(1);
+ serviceName = getRangerServiceName(clusterName);
+ }
+
+ if (StringUtils.equals(entityTypeName, ENTITY_TYPE_HIVE_DB)) {
+ if (components.size() > 2) {
+ dbName = components.get(2);
+ RangerPolicy.RangerPolicyResource dbPolicyResource = new RangerPolicy.RangerPolicyResource(dbName);
+ elements.put(RANGER_TYPE_HIVE_DB, dbPolicyResource);
+
+ } else {
+ LOG.error("invalid qualifiedName for HIVE_DB, qualifiedName=" + qualifiedName);
+ }
+ } else if (StringUtils.equals(entityTypeName, ENTITY_TYPE_HIVE_TABLE)) {
+ if (components.size() > 3) {
+ dbName = components.get(2);
+ tableName = components.get(3);
+ RangerPolicy.RangerPolicyResource dbPolicyResource = new RangerPolicy.RangerPolicyResource(dbName);
+ elements.put(RANGER_TYPE_HIVE_DB, dbPolicyResource);
+ RangerPolicy.RangerPolicyResource tablePolicyResource = new RangerPolicy.RangerPolicyResource(tableName);
+ elements.put(RANGER_TYPE_HIVE_TABLE, tablePolicyResource);
+ } else {
+ LOG.error("invalid qualifiedName for HIVE_TABLE, qualifiedName=" + qualifiedName);
+ }
+ } else if (StringUtils.equals(entityTypeName, ENTITY_TYPE_HIVE_COLUMN)) {
+ if (components.size() > 4) {
+ dbName = components.get(2);
+ tableName = components.get(3);
+ columnName = components.get(4);
+ RangerPolicy.RangerPolicyResource dbPolicyResource = new RangerPolicy.RangerPolicyResource(dbName);
+ elements.put(RANGER_TYPE_HIVE_DB, dbPolicyResource);
+ RangerPolicy.RangerPolicyResource tablePolicyResource = new RangerPolicy.RangerPolicyResource(tableName);
+ elements.put(RANGER_TYPE_HIVE_TABLE, tablePolicyResource);
+ RangerPolicy.RangerPolicyResource columnPolicyResource = new RangerPolicy.RangerPolicyResource(columnName);
+ elements.put(RANGER_TYPE_HIVE_COLUMN, columnPolicyResource);
+ } else {
+ LOG.error("invalid qualifiedName for HIVE_COLUMN, qualifiedName=" + qualifiedName);
+ }
+
+ }
+
+ RangerServiceResource ret = new RangerServiceResource();
+
+ ret.setGuid(entity.getId()._getId());
+ ret.setServiceName(serviceName);
+ ret.setResourceElements(elements);
+
+ return ret;
+ }
+
+ public String getRangerServiceName(String clusterName) {
+ String ret = getRangerServiceName(COMPONENT_NAME, clusterName);
+
+ if (StringUtils.isBlank(ret)) {
+ ret = clusterName + TagSyncConfig.TAGSYNC_DEFAULT_CLUSTERNAME_AND_COMPONENTNAME_SEPARATOR + COMPONENT_NAME;
+ }
+ return ret;
+ }
+
+ public final List<String> getQualifiedNameComponents(IReferenceableInstance entity) throws Exception {
+
+ String qualifiedNameAttributeName = getQualifiedNameAttributeName(entity.getTypeName());
+
+ String qualifiedName = getEntityAttribute(entity, qualifiedNameAttributeName, String.class);
+
+ List<String> ret = getQualifiedNameComponents(entity.getTypeName(), qualifiedName);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("----- Entity-Id:" + entity.getId()._getId());
+ LOG.debug("----- Entity-Type-Name:" + entity.getTypeName());
+ LOG.debug("----- Entity-Components -----");
+ int i = 0;
+ for (String value : ret) {
+ LOG.debug("----- Index:" + i++ + " Value:" + value);
+ }
+ }
+ return ret;
+ }
+
+ public final List<String> getQualifiedNameComponents(String entityTypeName, String qualifiedName) throws Exception {
+
+ String qualifiedNameAttributeName = getQualifiedNameAttributeName(entityTypeName);
+
+ if (StringUtils.isBlank(qualifiedName)) {
+ throw new Exception("Could not get a valid value for " + qualifiedNameAttributeName + " attribute from entity.");
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Received .... " + qualifiedNameAttributeName + "=" + qualifiedName + " for entity type " + entityTypeName);
+ }
+
+ String components[] = qualifiedName.split(clusterDelimiter);
+
+ if (components.length != 2) {
+ throw new Exception("Qualified Name does not contain cluster-name, qualifiedName=" + qualifiedName);
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("name-hierarchy=" + components[0] + ", cluster-name=" + components[1]);
+ }
+
+ String nameHierarchy[] = components[0].split(qualifiedNameDelimiter);
+
+ List<String> ret = new ArrayList<String>();
+
+ ret.add(qualifiedName);
+ ret.add(components[1]);
+
+ ret.addAll(Arrays.asList(nameHierarchy));
+
+ return ret;
+ }
+
+ public String getQualifiedNameAttributeName(String entityTypeName) {
+ return StringUtils.equals(entityTypeName, ENTITY_TYPE_HIVE_TABLE) ?
+ ENTITY_ATTRIBUTE_QUALIFIED_NAME_FOR_HIVE_TABLE : ENTITY_ATTRIBUTE_QUALIFIED_NAME;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasResourceMapper.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasResourceMapper.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasResourceMapper.java
new file mode 100644
index 0000000..fd94928
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasResourceMapper.java
@@ -0,0 +1,74 @@
+/*
+ * 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.ranger.tagsync.source.atlas;
+
+import org.apache.atlas.AtlasException;
+import org.apache.atlas.typesystem.IReferenceableInstance;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.ranger.plugin.model.RangerServiceResource;
+import org.apache.ranger.tagsync.process.TagSyncConfig;
+
+import java.util.Properties;
+import java.util.List;
+import java.util.Map;
+
+public abstract class AtlasResourceMapper {
+ private static final Log LOG = LogFactory.getLog(AtlasResourceMapper.class);
+
+ protected Properties properties;
+
+ public AtlasResourceMapper() {
+ }
+
+ public void initialize(Properties properties) {
+ this.properties = properties;
+ }
+
+ abstract public List<String> getSupportedEntityTypes();
+
+ abstract public RangerServiceResource buildResource(final IReferenceableInstance entity) throws Exception;
+
+
+ protected String getRangerServiceName(String componentName, String atlasInstanceName) {
+ String propName = TagSyncConfig.TAGSYNC_SERVICENAME_MAPPER_PROP_PREFIX + componentName
+ + ".instance." + atlasInstanceName
+ + TagSyncConfig.TAGSYNC_SERVICENAME_MAPPER_PROP_SUFFIX;
+
+ return properties.getProperty(propName);
+ }
+
+ static protected <T> T getEntityAttribute(IReferenceableInstance entity, String name, Class<T> type) {
+ T ret = null;
+
+ try {
+ Map<String, Object> valueMap = entity.getValuesMap();
+ ret = getAttribute(valueMap, name, type);
+ } catch (AtlasException exception) {
+ LOG.error("Cannot get map of values for entity: " + entity.getId()._getId(), exception);
+ }
+
+ return ret;
+ }
+
+ static protected <T> T getAttribute(Map<String, Object> map, String name, Class<T> type) {
+ return type.cast(map.get(name));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasResourceMapperUtil.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasResourceMapperUtil.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasResourceMapperUtil.java
new file mode 100644
index 0000000..f05d814
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasResourceMapperUtil.java
@@ -0,0 +1,124 @@
+/*
+ * 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 th
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.ranger.tagsync.source.atlas;
+
+import org.apache.atlas.typesystem.IReferenceableInstance;
+import org.apache.commons.lang.StringUtils;
+import org.apache.ranger.plugin.model.RangerServiceResource;
+
+import java.util.*;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.ranger.tagsync.process.TagSyncConfig;
+
+public class AtlasResourceMapperUtil {
+ private static final Log LOG = LogFactory.getLog(AtlasResourceMapperUtil.class);
+
+ private static Map<String, AtlasResourceMapper> atlasResourceMappers = new HashMap<String, AtlasResourceMapper>();
+
+ public static boolean isEntityTypeHandled(String entityTypeName) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> isEntityTypeHandled(entityTypeName=" + entityTypeName + ")");
+ }
+
+ AtlasResourceMapper mapper = atlasResourceMappers.get(entityTypeName);
+
+ boolean ret = mapper != null;
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== isEntityTypeHandled(entityTypeName=" + entityTypeName + ") : " + ret);
+ }
+
+ return ret;
+ }
+
+ public static RangerServiceResource getRangerServiceResource(IReferenceableInstance atlasEntity) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> getRangerServiceResource(" + atlasEntity.getId()._getId() +")");
+ }
+
+ RangerServiceResource resource = null;
+
+ AtlasResourceMapper mapper = atlasResourceMappers.get(atlasEntity.getTypeName());
+
+ if (mapper != null) {
+ try {
+ resource = mapper.buildResource(atlasEntity);
+ } catch (Exception exception) {
+ LOG.error("Could not get serviceResource for atlas entity:" + atlasEntity.getId()._getId() + ": ", exception);
+ }
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== getRangerServiceResource(" + atlasEntity.getId()._getId() +"): resource=" + resource);
+ }
+
+ return resource;
+ }
+
+ static public boolean initializeAtlasResourceMappers(Properties properties) {
+ final String MAPPER_NAME_DELIMIER = ",";
+
+ String customMapperNames = TagSyncConfig.getCustomAtlasResourceMappers(properties);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> initializeAtlasResourceMappers.initializeAtlasResourceMappers(" + customMapperNames + ")");
+ }
+ boolean ret = true;
+
+ String allResourceMappers = "org.apache.ranger.tagsync.source.atlas.AtlasHiveResourceMapper";
+
+ if (StringUtils.isNotBlank(customMapperNames)) {
+ allResourceMappers = allResourceMappers + MAPPER_NAME_DELIMIER + customMapperNames;
+ }
+
+ String[] mapperNamesArray = allResourceMappers.split(MAPPER_NAME_DELIMIER);
+
+ List<String> mapperNames = Arrays.asList(mapperNamesArray);
+
+ for (String mapperName : mapperNames) {
+ mapperName = mapperName.trim();
+ try {
+ Class clazz = Class.forName(mapperName);
+ AtlasResourceMapper resourceMapper = (AtlasResourceMapper) clazz.newInstance();
+
+ resourceMapper.initialize(properties);
+
+ for (String entityTypeName : resourceMapper.getSupportedEntityTypes()) {
+ add(entityTypeName, resourceMapper);
+ }
+
+ } catch (Exception exception) {
+ LOG.error("Failed to create AtlasResourceMapper:" + mapperName + ": ", exception);
+ ret = false;
+ }
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== initializeAtlasResourceMappers.initializeAtlasResourceMappers(" + allResourceMappers + "): " + ret);
+ }
+ return ret;
+ }
+
+ private static void add(String entityType, AtlasResourceMapper mapper) {
+ atlasResourceMappers.put(entityType, mapper);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasTagSource.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasTagSource.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasTagSource.java
new file mode 100644
index 0000000..7694b37
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/Atlas/AtlasTagSource.java
@@ -0,0 +1,197 @@
+/*
+ * 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.ranger.tagsync.source.atlas;
+
+
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provider;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import org.apache.atlas.notification.NotificationConsumer;
+import org.apache.atlas.notification.NotificationInterface;
+import org.apache.atlas.notification.NotificationModule;
+import org.apache.atlas.notification.entity.EntityNotification;
+
+import org.apache.ranger.tagsync.model.AbstractTagSource;
+import org.apache.ranger.plugin.util.ServiceTags;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.*;
+
+public class AtlasTagSource extends AbstractTagSource {
+ private static final Log LOG = LogFactory.getLog(AtlasTagSource.class);
+
+ public static final String TAGSYNC_ATLAS_PROPERTIES_FILE_NAME = "application.properties";
+
+ public static final String TAGSYNC_ATLAS_KAFKA_ENDPOINTS = "atlas.kafka.bootstrap.servers";
+ public static final String TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT = "atlas.kafka.zookeeper.connect";
+ public static final String TAGSYNC_ATLAS_CONSUMER_GROUP = "atlas.kafka.entities.group.id";
+
+ private ConsumerRunnable consumerTask;
+
+ @Override
+ public boolean initialize(Properties properties) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> AtlasTagSource.initialize()");
+ }
+
+ Properties atlasProperties = new Properties();
+
+ boolean ret = AtlasResourceMapperUtil.initializeAtlasResourceMappers(properties);
+
+ if (ret) {
+
+ InputStream inputStream = getClass().getClassLoader().getResourceAsStream(TAGSYNC_ATLAS_PROPERTIES_FILE_NAME);
+
+ if (inputStream != null) {
+ try {
+ atlasProperties.load(inputStream);
+ } catch (Exception exception) {
+ ret = false;
+ LOG.error("Cannot load Atlas application properties file, file-name:" + TAGSYNC_ATLAS_PROPERTIES_FILE_NAME, exception);
+ } finally {
+ try {
+ inputStream.close();
+ } catch (IOException ioException) {
+ LOG.error("Cannot close Atlas application properties file, file-name:\" + TAGSYNC_ATLAS_PROPERTIES_FILE_NAME", ioException);
+ }
+ }
+ } else {
+ ret = false;
+ LOG.error("Cannot find Atlas application properties file");
+ }
+ }
+
+ if (ret) {
+ if (StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_KAFKA_ENDPOINTS))) {
+ ret = false;
+ LOG.error("Value of property '" + TAGSYNC_ATLAS_KAFKA_ENDPOINTS + "' is not specified!");
+ }
+ if (StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT))) {
+ ret = false;
+ LOG.error("Value of property '" + TAGSYNC_ATLAS_ZOOKEEPER_ENDPOINT + "' is not specified!");
+ }
+ if (StringUtils.isBlank(atlasProperties.getProperty(TAGSYNC_ATLAS_CONSUMER_GROUP))) {
+ ret = false;
+ LOG.error("Value of property '" + TAGSYNC_ATLAS_CONSUMER_GROUP + "' is not specified!");
+ }
+ }
+
+ if (ret) {
+ NotificationModule notificationModule = new NotificationModule();
+
+ Injector injector = Guice.createInjector(notificationModule);
+
+ Provider<NotificationInterface> consumerProvider = injector.getProvider(NotificationInterface.class);
+ NotificationInterface notification = consumerProvider.get();
+ List<NotificationConsumer<EntityNotification>> iterators = notification.createConsumers(NotificationInterface.NotificationType.ENTITIES, 1);
+
+ consumerTask = new ConsumerRunnable(iterators.get(0));
+
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== AtlasTagSource.initialize(), result=" + ret);
+ }
+ return ret;
+ }
+
+ @Override
+ public boolean start() {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> AtlasTagSource.start()");
+ }
+ Thread consumerThread = null;
+ if (consumerTask == null) {
+ LOG.error("No consumerTask!!!");
+ } else {
+ consumerThread = new Thread(consumerTask);
+ consumerThread.setDaemon(true);
+ consumerThread.start();
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== AtlasTagSource.start()");
+ }
+ return consumerThread != null;
+ }
+
+ @Override
+ public boolean isChanged() {
+ return true;
+ }
+
+ private static void printEntityNotification(EntityNotification notification) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Notification-Type: " + notification.getOperationType());
+ AtlasEntityWithTraits entityWithTraits = new AtlasEntityWithTraits(notification.getEntity(), notification.getAllTraits());
+ LOG.debug(entityWithTraits);
+ }
+ }
+
+ private class ConsumerRunnable implements Runnable {
+
+ private final NotificationConsumer<EntityNotification> consumer;
+
+ private ConsumerRunnable(NotificationConsumer<EntityNotification> consumer) {
+ this.consumer = consumer;
+ }
+
+ private boolean hasNext() {
+ boolean ret = false;
+ try {
+ ret = consumer.hasNext();
+ } catch (Exception exception) {
+ LOG.error("EntityNotification consumer threw exception, IGNORING...:", exception);
+ }
+ return ret;
+ }
+
+ @Override
+ public void run() {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> ConsumerRunnable.run()");
+ }
+ while (!shutdown) {
+ if (hasNext()) {
+ EntityNotification notification = consumer.next();
+ if (notification != null) {
+ printEntityNotification(notification);
+
+ ServiceTags serviceTags = AtlasNotificationMapper.processEntityNotification(notification);
+ if (serviceTags == null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Did not create ServiceTags structure for notification type:" + notification.getOperationType());
+ }
+ } else {
+ updateSink(serviceTags);
+ }
+ }
+ }
+ }
+ LOG.info("Shutting down the Tag-Atlas-source thread");
+ }
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-ranger/blob/5b9c094f/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasEntityWithTraits.java
----------------------------------------------------------------------
diff --git a/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasEntityWithTraits.java b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasEntityWithTraits.java
new file mode 100644
index 0000000..e6a4b36
--- /dev/null
+++ b/tagsync/src/main/java/org/apache/ranger/tagsync/source/atlas/AtlasEntityWithTraits.java
@@ -0,0 +1,96 @@
+/*
+ * 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.ranger.tagsync.source.atlas;
+
+import org.apache.atlas.AtlasException;
+import org.apache.atlas.typesystem.IReferenceableInstance;
+import org.apache.atlas.typesystem.IStruct;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+
+public class AtlasEntityWithTraits {
+
+ private final IReferenceableInstance entity;
+ private final List<IStruct> traits;
+
+ public AtlasEntityWithTraits(IReferenceableInstance entity, List<IStruct> traits) {
+ this.entity = entity;
+ this.traits = traits;
+ }
+
+ public IReferenceableInstance getEntity() {
+ return entity;
+ }
+
+ public List<IStruct> getAllTraits() {
+ return traits == null ? new LinkedList<IStruct>() : traits;
+ }
+
+ @Override
+ public String toString( ) {
+ StringBuilder sb = new StringBuilder();
+
+ toString(sb);
+
+ return sb.toString();
+ }
+
+ public void toString(StringBuilder sb) {
+
+ sb.append("AtlasEntityWithTraits={ ");
+
+ sb.append("Entity-Id: " + entity.getId()._getId()).append(", ")
+ .append("Entity-Type: " + entity.getTypeName()).append(", ");
+
+ sb.append("Entity-Values={ ");
+ try {
+ for (Map.Entry<String, Object> entry : entity.getValuesMap().entrySet()) {
+ sb.append("{").append(entry.getKey()).append(", ").append(entry.getValue()).append("}, ");
+ }
+ } catch (AtlasException exception) {
+ // Ignore
+ }
+ sb.append(" }");
+
+ sb.append(", Entity-Traits={ ");
+ for (IStruct trait : traits) {
+ try {
+ sb.append("{traitType=").append(trait.getTypeName()).append(", ");
+ Map<String, Object> traitValues = trait.getValuesMap();
+ sb.append("{");
+ for (Map.Entry<String, Object> valueEntry : traitValues.entrySet()) {
+ sb.append("{").append(valueEntry.getKey()).append(", ").append(valueEntry.getValue()).append("}");
+ }
+ sb.append("}");
+
+ sb.append(" }");
+ } catch (AtlasException exception) {
+ // Ignore
+ }
+ }
+ sb.append(" }");
+
+ sb.append(" }");
+
+ }
+
+}