You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@atlas.apache.org by sa...@apache.org on 2018/05/24 04:50:00 UTC
atlas git commit: ATLAS-2715: Create audit events for term-entity
association and disassociation
Repository: atlas
Updated Branches:
refs/heads/master bcd5bb607 -> 1deb8c16e
ATLAS-2715: Create audit events for term-entity association and disassociation
Project: http://git-wip-us.apache.org/repos/asf/atlas/repo
Commit: http://git-wip-us.apache.org/repos/asf/atlas/commit/1deb8c16
Tree: http://git-wip-us.apache.org/repos/asf/atlas/tree/1deb8c16
Diff: http://git-wip-us.apache.org/repos/asf/atlas/diff/1deb8c16
Branch: refs/heads/master
Commit: 1deb8c16ec8eca68962eff7f8e4d6dac81fd37aa
Parents: bcd5bb6
Author: Sarath Subramanian <ss...@hortonworks.com>
Authored: Wed May 23 15:02:36 2018 -0700
Committer: Sarath Subramanian <ss...@hortonworks.com>
Committed: Wed May 23 21:38:03 2018 -0700
----------------------------------------------------------------------
.../java/org/apache/atlas/EntityAuditEvent.java | 7 +-
dashboardv2/public/js/utils/Enums.js | 4 +-
.../atlas/listener/EntityChangeListenerV2.java | 18 ++++++
.../atlas/model/audit/EntityAuditEventV2.java | 7 +-
.../atlas/model/glossary/AtlasGlossaryTerm.java | 11 ++++
.../apache/atlas/glossary/GlossaryService.java | 41 +++++++-----
.../repository/audit/EntityAuditListener.java | 38 +++++++++++
.../repository/audit/EntityAuditListenerV2.java | 67 ++++++++++++++++++--
.../audit/HBaseBasedAuditRepository.java | 28 +++++---
.../graph/v1/AtlasEntityChangeNotifier.java | 54 ++++++++++++++++
.../atlas/listener/EntityChangeListener.java | 19 ++++++
.../NoOpNotificationChangeListener.java | 11 ++++
.../EntityNotificationListenerV2.java | 12 ++++
.../NotificationEntityChangeListener.java | 11 ++++
14 files changed, 294 insertions(+), 34 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/client/client-v1/src/main/java/org/apache/atlas/EntityAuditEvent.java
----------------------------------------------------------------------
diff --git a/client/client-v1/src/main/java/org/apache/atlas/EntityAuditEvent.java b/client/client-v1/src/main/java/org/apache/atlas/EntityAuditEvent.java
index 7077e15..fcd6a62 100644
--- a/client/client-v1/src/main/java/org/apache/atlas/EntityAuditEvent.java
+++ b/client/client-v1/src/main/java/org/apache/atlas/EntityAuditEvent.java
@@ -48,7 +48,8 @@ public class EntityAuditEvent implements Serializable {
public enum EntityAuditAction {
ENTITY_CREATE, ENTITY_UPDATE, ENTITY_DELETE, TAG_ADD, TAG_DELETE, TAG_UPDATE,
PROPAGATED_TAG_ADD, PROPAGATED_TAG_DELETE, PROPAGATED_TAG_UPDATE,
- ENTITY_IMPORT_CREATE, ENTITY_IMPORT_UPDATE, ENTITY_IMPORT_DELETE;
+ ENTITY_IMPORT_CREATE, ENTITY_IMPORT_UPDATE, ENTITY_IMPORT_DELETE,
+ TERM_ADD, TERM_DELETE;
public static EntityAuditAction fromString(String strValue) {
switch (strValue) {
@@ -79,6 +80,10 @@ public class EntityAuditEvent implements Serializable {
return PROPAGATED_TAG_DELETE;
case "PROPAGATED_TAG_UPDATE":
return PROPAGATED_TAG_UPDATE;
+ case "TERM_ADD":
+ return TERM_ADD;
+ case "TERM_DELETE":
+ return TERM_DELETE;
}
throw new IllegalArgumentException("No enum constant " + EntityAuditAction.class.getCanonicalName() + "." + strValue);
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/dashboardv2/public/js/utils/Enums.js
----------------------------------------------------------------------
diff --git a/dashboardv2/public/js/utils/Enums.js b/dashboardv2/public/js/utils/Enums.js
index 1e493e5..81de527 100644
--- a/dashboardv2/public/js/utils/Enums.js
+++ b/dashboardv2/public/js/utils/Enums.js
@@ -33,7 +33,9 @@ define(['require'], function(require) {
PROPAGATED_CLASSIFICATION_UPDATE: "Propagated Classification Updated",
ENTITY_IMPORT_CREATE: "Entity Created by import",
ENTITY_IMPORT_UPDATE: "Entity Updated by import",
- ENTITY_IMPORT_DELETE: "Entity Deleted by import"
+ ENTITY_IMPORT_DELETE: "Entity Deleted by import",
+ TERM_ADD: "Term Added",
+ TERM_DELETE: "Term Deleted"
}
Enums.entityStateReadOnly = {
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/intg/src/main/java/org/apache/atlas/listener/EntityChangeListenerV2.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/listener/EntityChangeListenerV2.java b/intg/src/main/java/org/apache/atlas/listener/EntityChangeListenerV2.java
index 9c735a0..cccf0d4 100644
--- a/intg/src/main/java/org/apache/atlas/listener/EntityChangeListenerV2.java
+++ b/intg/src/main/java/org/apache/atlas/listener/EntityChangeListenerV2.java
@@ -19,8 +19,10 @@
package org.apache.atlas.listener;
import org.apache.atlas.exception.AtlasBaseException;
+import org.apache.atlas.model.glossary.AtlasGlossaryTerm;
import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntity;
+import org.apache.atlas.model.instance.AtlasRelatedObjectId;
import java.util.List;
@@ -78,4 +80,20 @@ public interface EntityChangeListenerV2 {
* @throws AtlasBaseException if the listener notification fails
*/
void onClassificationsDeleted(AtlasEntity entity, List<AtlasClassification> classifications) throws AtlasBaseException;
+
+ /**
+ * This is upon adding a new term to an entity.
+ *
+ * @param term the term
+ * @param entities list of entities to which the term is assigned
+ */
+ void onTermAdded(AtlasGlossaryTerm term, List<AtlasRelatedObjectId> entities) throws AtlasBaseException;
+
+ /**
+ * This is upon removing a term from an entity.
+ *
+ * @param term the term
+ * @param entities list of entities to which the term is assigned
+ */
+ void onTermDeleted(AtlasGlossaryTerm term, List<AtlasRelatedObjectId> entities) throws AtlasBaseException;
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/intg/src/main/java/org/apache/atlas/model/audit/EntityAuditEventV2.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/model/audit/EntityAuditEventV2.java b/intg/src/main/java/org/apache/atlas/model/audit/EntityAuditEventV2.java
index 787f5a9..649f11f 100644
--- a/intg/src/main/java/org/apache/atlas/model/audit/EntityAuditEventV2.java
+++ b/intg/src/main/java/org/apache/atlas/model/audit/EntityAuditEventV2.java
@@ -49,7 +49,8 @@ public class EntityAuditEventV2 implements Serializable {
ENTITY_CREATE, ENTITY_UPDATE, ENTITY_DELETE,
ENTITY_IMPORT_CREATE, ENTITY_IMPORT_UPDATE, ENTITY_IMPORT_DELETE,
CLASSIFICATION_ADD, CLASSIFICATION_DELETE, CLASSIFICATION_UPDATE,
- PROPAGATED_CLASSIFICATION_ADD, PROPAGATED_CLASSIFICATION_DELETE, PROPAGATED_CLASSIFICATION_UPDATE;
+ PROPAGATED_CLASSIFICATION_ADD, PROPAGATED_CLASSIFICATION_DELETE, PROPAGATED_CLASSIFICATION_UPDATE,
+ TERM_ADD, TERM_DELETE;
public static EntityAuditActionV2 fromString(String strValue) {
switch (strValue) {
@@ -80,6 +81,10 @@ public class EntityAuditEventV2 implements Serializable {
return PROPAGATED_CLASSIFICATION_DELETE;
case "PROPAGATED_CLASSIFICATION_UPDATE":
return PROPAGATED_CLASSIFICATION_UPDATE;
+ case "TERM_ADD":
+ return TERM_ADD;
+ case "TERM_DELETE":
+ return TERM_DELETE;
}
throw new IllegalArgumentException("No enum constant " + EntityAuditActionV2.class.getCanonicalName() + "." + strValue);
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/intg/src/main/java/org/apache/atlas/model/glossary/AtlasGlossaryTerm.java
----------------------------------------------------------------------
diff --git a/intg/src/main/java/org/apache/atlas/model/glossary/AtlasGlossaryTerm.java b/intg/src/main/java/org/apache/atlas/model/glossary/AtlasGlossaryTerm.java
index daf6c7b..578d4de 100644
--- a/intg/src/main/java/org/apache/atlas/model/glossary/AtlasGlossaryTerm.java
+++ b/intg/src/main/java/org/apache/atlas/model/glossary/AtlasGlossaryTerm.java
@@ -24,6 +24,7 @@ import org.apache.atlas.model.glossary.relations.AtlasGlossaryHeader;
import org.apache.atlas.model.glossary.relations.AtlasRelatedTermHeader;
import org.apache.atlas.model.glossary.relations.AtlasTermCategorizationHeader;
import org.apache.atlas.model.instance.AtlasRelatedObjectId;
+import org.apache.atlas.type.AtlasType;
import org.apache.commons.collections.CollectionUtils;
import java.util.HashMap;
@@ -286,6 +287,16 @@ public class AtlasGlossaryTerm extends AtlasGlossaryBaseObject {
}
@JsonIgnore
+ public String toAuditString() {
+ AtlasGlossaryTerm t = new AtlasGlossaryTerm();
+ t.setGuid(this.getGuid());
+ t.setDisplayName(this.getDisplayName());
+ t.setQualifiedName(this.getQualifiedName());
+
+ return AtlasType.toJson(t);
+ }
+
+ @JsonIgnore
public boolean hasTerms() {
return hasTerms;
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/repository/src/main/java/org/apache/atlas/glossary/GlossaryService.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/glossary/GlossaryService.java b/repository/src/main/java/org/apache/atlas/glossary/GlossaryService.java
index a5499e0..3467259 100644
--- a/repository/src/main/java/org/apache/atlas/glossary/GlossaryService.java
+++ b/repository/src/main/java/org/apache/atlas/glossary/GlossaryService.java
@@ -31,6 +31,7 @@ import org.apache.atlas.model.instance.AtlasRelatedObjectId;
import org.apache.atlas.repository.graphdb.AtlasVertex;
import org.apache.atlas.repository.ogm.DataAccess;
import org.apache.atlas.repository.store.graph.AtlasRelationshipStore;
+import org.apache.atlas.repository.store.graph.v1.AtlasEntityChangeNotifier;
import org.apache.atlas.repository.store.graph.v1.AtlasGraphUtilsV1;
import org.apache.atlas.type.AtlasTypeRegistry;
import org.apache.commons.collections.CollectionUtils;
@@ -56,24 +57,26 @@ import static org.apache.atlas.glossary.GlossaryUtils.getGlossarySkeleton;
@Service
public class GlossaryService {
- private static final Logger LOG = LoggerFactory.getLogger(GlossaryService.class);
- private static final boolean DEBUG_ENABLED = LOG.isDebugEnabled();
+ private static final Logger LOG = LoggerFactory.getLogger(GlossaryService.class);
+ private static final boolean DEBUG_ENABLED = LOG.isDebugEnabled();
+ private static final String QUALIFIED_NAME_ATTR = "qualifiedName";
- private static final String QUALIFIED_NAME_ATTR = "qualifiedName";
-
- private final DataAccess dataAccess;
- private final GlossaryTermUtils glossaryTermUtils;
- private final GlossaryCategoryUtils glossaryCategoryUtils;
- private final AtlasTypeRegistry atlasTypeRegistry;
+ private final DataAccess dataAccess;
+ private final GlossaryTermUtils glossaryTermUtils;
+ private final GlossaryCategoryUtils glossaryCategoryUtils;
+ private final AtlasTypeRegistry atlasTypeRegistry;
+ private final AtlasEntityChangeNotifier entityChangeNotifier;
private final char[] invalidNameChars = {'@', '.'};
@Inject
- public GlossaryService(DataAccess dataAccess, final AtlasRelationshipStore relationshipStore, final AtlasTypeRegistry typeRegistry) {
- this.dataAccess = dataAccess;
- this.atlasTypeRegistry = typeRegistry;
- glossaryTermUtils = new GlossaryTermUtils(relationshipStore, typeRegistry, dataAccess);
- glossaryCategoryUtils = new GlossaryCategoryUtils(relationshipStore, typeRegistry, dataAccess);
+ public GlossaryService(DataAccess dataAccess, final AtlasRelationshipStore relationshipStore,
+ final AtlasTypeRegistry typeRegistry, AtlasEntityChangeNotifier entityChangeNotifier) {
+ this.dataAccess = dataAccess;
+ atlasTypeRegistry = typeRegistry;
+ glossaryTermUtils = new GlossaryTermUtils(relationshipStore, typeRegistry, dataAccess);
+ glossaryCategoryUtils = new GlossaryCategoryUtils(relationshipStore, typeRegistry, dataAccess);
+ this.entityChangeNotifier = entityChangeNotifier;
}
/**
@@ -477,24 +480,32 @@ public class GlossaryService {
if (DEBUG_ENABLED) {
LOG.debug("==> GlossaryService.assignTermToEntities({}, {})", termGuid, relatedObjectIds);
}
+
AtlasGlossaryTerm glossaryTerm = dataAccess.load(getAtlasGlossaryTermSkeleton(termGuid));
+
glossaryTermUtils.processTermAssignments(glossaryTerm, relatedObjectIds);
+ entityChangeNotifier.onTermAddedToEntities(glossaryTerm, relatedObjectIds);
+
if (DEBUG_ENABLED) {
LOG.debug("<== GlossaryService.assignTermToEntities()");
}
+
}
@GraphTransaction
public void removeTermFromEntities(String termGuid, List<AtlasRelatedObjectId> relatedObjectIds) throws AtlasBaseException {
- if (LOG.isDebugEnabled()) {
+ if (DEBUG_ENABLED) {
LOG.debug("==> GlossaryService.removeTermFromEntities({}, {})", termGuid, relatedObjectIds);
}
AtlasGlossaryTerm glossaryTerm = dataAccess.load(getAtlasGlossaryTermSkeleton(termGuid));
+
glossaryTermUtils.processTermDissociation(glossaryTerm, relatedObjectIds);
- if (LOG.isDebugEnabled()) {
+ entityChangeNotifier.onTermDeletedFromEntities(glossaryTerm, relatedObjectIds);
+
+ if (DEBUG_ENABLED) {
LOG.debug("<== GlossaryService.removeTermFromEntities()");
}
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListener.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListener.java b/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListener.java
index a085e8e..7369c5e 100644
--- a/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListener.java
+++ b/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListener.java
@@ -23,6 +23,7 @@ import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.EntityAuditEvent.EntityAuditAction;
import org.apache.atlas.RequestContextV1;
import org.apache.atlas.listener.EntityChangeListener;
+import org.apache.atlas.model.glossary.AtlasGlossaryTerm;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.v1.model.instance.Struct;
import org.apache.atlas.type.AtlasEntityType;
@@ -43,6 +44,9 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static org.apache.atlas.EntityAuditEvent.EntityAuditAction.TERM_ADD;
+import static org.apache.atlas.EntityAuditEvent.EntityAuditAction.TERM_DELETE;
+
/**
* Listener on entity create/update/delete, tag add/delete. Adds the corresponding audit event to the audit repository.
*/
@@ -127,6 +131,28 @@ public class EntityAuditListener implements EntityChangeListener {
auditRepository.putEventsV1(events);
}
+ @Override
+ public void onTermAdded(Collection<Referenceable> entities, AtlasGlossaryTerm term) throws AtlasException {
+ List<EntityAuditEvent> events = new ArrayList<>();
+
+ for (Referenceable entity : entities) {
+ events.add(createEvent(entity, TERM_ADD, "Added term: " + term.toAuditString()));
+ }
+
+ auditRepository.putEventsV1(events);
+ }
+
+ @Override
+ public void onTermDeleted(Collection<Referenceable> entities, AtlasGlossaryTerm term) throws AtlasException {
+ List<EntityAuditEvent> events = new ArrayList<>();
+
+ for (Referenceable entity : entities) {
+ events.add(createEvent(entity, TERM_DELETE, "Deleted term: " + term.toAuditString()));
+ }
+
+ auditRepository.putEventsV1(events);
+ }
+
public List<EntityAuditEvent> getAuditEvents(String guid) throws AtlasException{
return auditRepository.listEventsV1(guid, null, (short) 10);
}
@@ -290,6 +316,12 @@ public class EntityAuditListener implements EntityChangeListener {
case ENTITY_IMPORT_DELETE:
ret = "Deleted by import: ";
break;
+ case TERM_ADD:
+ ret = "Added term: ";
+ break;
+ case TERM_DELETE:
+ ret = "Deleted term: ";
+ break;
default:
ret = "Unknown: ";
}
@@ -328,6 +360,12 @@ public class EntityAuditListener implements EntityChangeListener {
case ENTITY_IMPORT_DELETE:
ret = "Deleted by import: ";
break;
+ case TERM_ADD:
+ ret = "Added term: ";
+ break;
+ case TERM_DELETE:
+ ret = "Deleted term: ";
+ break;
default:
ret = "Unknown: ";
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListenerV2.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListenerV2.java b/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListenerV2.java
index 970b14a..a988e47 100644
--- a/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListenerV2.java
+++ b/repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListenerV2.java
@@ -17,15 +17,18 @@
*/
package org.apache.atlas.repository.audit;
-import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.EntityAuditEvent.EntityAuditAction;
import org.apache.atlas.RequestContextV1;
import org.apache.atlas.model.audit.EntityAuditEventV2;
import org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditActionV2;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.listener.EntityChangeListenerV2;
+import org.apache.atlas.model.glossary.AtlasGlossaryTerm;
import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntity;
+import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
+import org.apache.atlas.model.instance.AtlasRelatedObjectId;
+import org.apache.atlas.repository.converters.AtlasInstanceConverter;
import org.apache.atlas.type.AtlasEntityType;
import org.apache.atlas.type.AtlasStructType.AtlasAttribute;
import org.apache.atlas.type.AtlasType;
@@ -56,18 +59,22 @@ import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditActionV
import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditActionV2.PROPAGATED_CLASSIFICATION_ADD;
import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditActionV2.PROPAGATED_CLASSIFICATION_DELETE;
import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditActionV2.PROPAGATED_CLASSIFICATION_UPDATE;
+import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditActionV2.TERM_ADD;
+import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditActionV2.TERM_DELETE;
@Component
public class EntityAuditListenerV2 implements EntityChangeListenerV2 {
private static final Logger LOG = LoggerFactory.getLogger(EntityAuditListenerV2.class);
- private final EntityAuditRepository auditRepository;
- private final AtlasTypeRegistry typeRegistry;
+ private final EntityAuditRepository auditRepository;
+ private final AtlasTypeRegistry typeRegistry;
+ private final AtlasInstanceConverter instanceConverter;
@Inject
- public EntityAuditListenerV2(EntityAuditRepository auditRepository, AtlasTypeRegistry typeRegistry) {
- this.auditRepository = auditRepository;
- this.typeRegistry = typeRegistry;
+ public EntityAuditListenerV2(EntityAuditRepository auditRepository, AtlasTypeRegistry typeRegistry, AtlasInstanceConverter instanceConverter) {
+ this.auditRepository = auditRepository;
+ this.typeRegistry = typeRegistry;
+ this.instanceConverter = instanceConverter;
}
@Override
@@ -167,6 +174,42 @@ public class EntityAuditListenerV2 implements EntityChangeListenerV2 {
}
}
+ @Override
+ public void onTermAdded(AtlasGlossaryTerm term, List<AtlasRelatedObjectId> entities) throws AtlasBaseException {
+ if (term != null && CollectionUtils.isNotEmpty(entities)) {
+ List<EntityAuditEventV2> events = new ArrayList<>();
+
+ for (AtlasRelatedObjectId relatedObjectId : entities) {
+ AtlasEntityWithExtInfo entityWithExtInfo = instanceConverter.getAndCacheEntity(relatedObjectId.getGuid());
+ AtlasEntity entity = (entityWithExtInfo != null) ? entityWithExtInfo.getEntity() : null;
+
+ if (entity != null) {
+ events.add(createEvent(entity, TERM_ADD, "Added term: " + term.toAuditString()));
+ }
+ }
+
+ auditRepository.putEventsV2(events);
+ }
+ }
+
+ @Override
+ public void onTermDeleted(AtlasGlossaryTerm term, List<AtlasRelatedObjectId> entities) throws AtlasBaseException {
+ if (term != null && CollectionUtils.isNotEmpty(entities)) {
+ List<EntityAuditEventV2> events = new ArrayList<>();
+
+ for (AtlasRelatedObjectId relatedObjectId : entities) {
+ AtlasEntityWithExtInfo entityWithExtInfo = instanceConverter.getAndCacheEntity(relatedObjectId.getGuid());
+ AtlasEntity entity = (entityWithExtInfo != null) ? entityWithExtInfo.getEntity() : null;
+
+ if (entity != null) {
+ events.add(createEvent(entity, TERM_DELETE, "Deleted term: " + term.toAuditString()));
+ }
+ }
+
+ auditRepository.putEventsV2(events);
+ }
+ }
+
private EntityAuditEventV2 createEvent(AtlasEntity entity, EntityAuditActionV2 action, String details) {
return new EntityAuditEventV2(entity.getGuid(), RequestContextV1.get().getRequestTime(),
RequestContextV1.get().getUser(), action, details, entity);
@@ -310,6 +353,12 @@ public class EntityAuditListenerV2 implements EntityChangeListenerV2 {
case ENTITY_IMPORT_DELETE:
ret = "Deleted by import: ";
break;
+ case TERM_ADD:
+ ret = "Added term: ";
+ break;
+ case TERM_DELETE:
+ ret = "Deleted term: ";
+ break;
default:
ret = "Unknown: ";
}
@@ -348,6 +397,12 @@ public class EntityAuditListenerV2 implements EntityChangeListenerV2 {
case ENTITY_IMPORT_DELETE:
ret = "Deleted by import: ";
break;
+ case TERM_ADD:
+ ret = "Added term: ";
+ break;
+ case TERM_DELETE:
+ ret = "Deleted term: ";
+ break;
default:
ret = "Unknown: ";
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java b/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java
index 3656b85..65686ea 100644
--- a/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java
+++ b/repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java
@@ -70,6 +70,8 @@ import java.util.List;
import static org.apache.atlas.EntityAuditEvent.EntityAuditAction.TAG_ADD;
import static org.apache.atlas.EntityAuditEvent.EntityAuditAction.TAG_DELETE;
import static org.apache.atlas.EntityAuditEvent.EntityAuditAction.TAG_UPDATE;
+import static org.apache.atlas.EntityAuditEvent.EntityAuditAction.TERM_ADD;
+import static org.apache.atlas.EntityAuditEvent.EntityAuditAction.TERM_DELETE;
import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditType;
import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditType.ENTITY_AUDIT_V1;
import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditType.ENTITY_AUDIT_V2;
@@ -319,19 +321,25 @@ public class HBaseBasedAuditRepository extends AbstractStorageBasedAuditReposito
if (StringUtils.isNotEmpty(v1DetailsWithPrefix)) {
EntityAuditAction v1AuditAction = EntityAuditAction.fromString(getResultString(result, COLUMN_ACTION));
- String v1AuditPrefix = EntityAuditListener.getV1AuditPrefix(v1AuditAction);
- String[] split = v1DetailsWithPrefix.split(v1AuditPrefix);
- if (ArrayUtils.isNotEmpty(split) && split.length == 2) {
- String v1AuditDetails = split[1];
- Referenceable referenceable = AtlasType.fromV1Json(v1AuditDetails, Referenceable.class);
- String v2Json = (referenceable != null) ? toV2Json(referenceable, v1AuditAction) : v1AuditDetails;
+ if (v1AuditAction == TERM_ADD || v1AuditAction == TERM_DELETE) {
+ // for terms audit v1 and v2 structure is same
+ ret = v1DetailsWithPrefix;
+ } else {
+ String v1AuditPrefix = EntityAuditListener.getV1AuditPrefix(v1AuditAction);
+ String[] split = v1DetailsWithPrefix.split(v1AuditPrefix);
+
+ if (ArrayUtils.isNotEmpty(split) && split.length == 2) {
+ String v1AuditDetails = split[1];
+ Referenceable referenceable = AtlasType.fromV1Json(v1AuditDetails, Referenceable.class);
+ String v2Json = (referenceable != null) ? toV2Json(referenceable, v1AuditAction) : v1AuditDetails;
- if (v2Json != null) {
- ret = getV2AuditPrefix(v1AuditAction) + v2Json;
+ if (v2Json != null) {
+ ret = getV2AuditPrefix(v1AuditAction) + v2Json;
+ }
+ } else {
+ ret = v1DetailsWithPrefix;
}
- } else {
- ret = v1DetailsWithPrefix;
}
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/repository/src/main/java/org/apache/atlas/repository/store/graph/v1/AtlasEntityChangeNotifier.java
----------------------------------------------------------------------
diff --git a/repository/src/main/java/org/apache/atlas/repository/store/graph/v1/AtlasEntityChangeNotifier.java b/repository/src/main/java/org/apache/atlas/repository/store/graph/v1/AtlasEntityChangeNotifier.java
index 083600e..0e90336 100644
--- a/repository/src/main/java/org/apache/atlas/repository/store/graph/v1/AtlasEntityChangeNotifier.java
+++ b/repository/src/main/java/org/apache/atlas/repository/store/graph/v1/AtlasEntityChangeNotifier.java
@@ -25,11 +25,13 @@ import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.listener.EntityChangeListener;
import org.apache.atlas.listener.EntityChangeListenerV2;
import org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditActionV2;
+import org.apache.atlas.model.glossary.AtlasGlossaryTerm;
import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
import org.apache.atlas.model.instance.AtlasEntityHeader;
+import org.apache.atlas.model.instance.AtlasRelatedObjectId;
import org.apache.atlas.model.instance.EntityMutationResponse;
import org.apache.atlas.model.instance.EntityMutations.EntityOperation;
import org.apache.atlas.type.AtlasEntityType;
@@ -192,6 +194,44 @@ public class AtlasEntityChangeNotifier {
}
}
+ public void onTermAddedToEntities(AtlasGlossaryTerm term, List<AtlasRelatedObjectId> entityIds) throws AtlasBaseException {
+ // listeners notified on term-entity association only if v2 notifications are enabled
+ if (isV2EntityNotificationEnabled()) {
+ for (EntityChangeListenerV2 listener : entityChangeListenersV2) {
+ listener.onTermAdded(term, entityIds);
+ }
+ } else {
+ List<Referenceable> entityRefs = toReferenceables(entityIds);
+
+ for (EntityChangeListener listener : entityChangeListeners) {
+ try {
+ listener.onTermAdded(entityRefs, term);
+ } catch (AtlasException e) {
+ throw new AtlasBaseException(AtlasErrorCode.NOTIFICATION_FAILED, e, getListenerName(listener), "TermAdd");
+ }
+ }
+ }
+ }
+
+ public void onTermDeletedFromEntities(AtlasGlossaryTerm term, List<AtlasRelatedObjectId> entityIds) throws AtlasBaseException {
+ // listeners notified on term-entity disassociation only if v2 notifications are enabled
+ if (isV2EntityNotificationEnabled()) {
+ for (EntityChangeListenerV2 listener : entityChangeListenersV2) {
+ listener.onTermDeleted(term, entityIds);
+ }
+ } else {
+ List<Referenceable> entityRefs = toReferenceables(entityIds);
+
+ for (EntityChangeListener listener : entityChangeListeners) {
+ try {
+ listener.onTermDeleted(entityRefs, term);
+ } catch (AtlasException e) {
+ throw new AtlasBaseException(AtlasErrorCode.NOTIFICATION_FAILED, e, getListenerName(listener), "TermDelete");
+ }
+ }
+ }
+ }
+
public void notifyPropagatedEntities() throws AtlasBaseException {
RequestContextV1 context = RequestContextV1.get();
Map<String, List<AtlasClassification>> addedPropagations = context.getAddedPropagations();
@@ -297,6 +337,20 @@ public class AtlasEntityChangeNotifier {
return ret;
}
+ private List<Referenceable> toReferenceables(List<AtlasRelatedObjectId> entityIds) throws AtlasBaseException {
+ List<Referenceable> ret = new ArrayList<>();
+
+ if (CollectionUtils.isNotEmpty(entityIds)) {
+ for (AtlasRelatedObjectId relatedObjectId : entityIds) {
+ String entityGuid = relatedObjectId.getGuid();
+
+ ret.add(toReferenceable(entityGuid));
+ }
+ }
+
+ return ret;
+ }
+
private Referenceable toReferenceable(String entityId) throws AtlasBaseException {
Referenceable ret = null;
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/server-api/src/main/java/org/apache/atlas/listener/EntityChangeListener.java
----------------------------------------------------------------------
diff --git a/server-api/src/main/java/org/apache/atlas/listener/EntityChangeListener.java b/server-api/src/main/java/org/apache/atlas/listener/EntityChangeListener.java
index ab379e0..e24f582 100644
--- a/server-api/src/main/java/org/apache/atlas/listener/EntityChangeListener.java
+++ b/server-api/src/main/java/org/apache/atlas/listener/EntityChangeListener.java
@@ -19,6 +19,7 @@
package org.apache.atlas.listener;
import org.apache.atlas.AtlasException;
+import org.apache.atlas.model.glossary.AtlasGlossaryTerm;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.v1.model.instance.Struct;
@@ -86,4 +87,22 @@ public interface EntityChangeListener {
* @throws AtlasException
*/
void onEntitiesDeleted(Collection<Referenceable> entities, boolean isImport) throws AtlasException;
+
+ /**
+ * This is upon adding a new term to a list of typed instance.
+ *
+ * @param entities entity list
+ * @param term term that needs to be added to entity
+ * @throws AtlasException if the listener notification fails
+ */
+ void onTermAdded(Collection<Referenceable> entities, AtlasGlossaryTerm term) throws AtlasException;
+
+ /**
+ * This is upon adding a new trait to a typed instance.
+ *
+ * @param entities entity list
+ * @param term term that needs to be added to entity
+ * @throws AtlasException if the listener notification fails
+ */
+ void onTermDeleted(Collection<Referenceable> entities, AtlasGlossaryTerm term) throws AtlasException;
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/tools/atlas-migration-exporter/src/main/java/org/apache/atlas/migration/NoOpNotificationChangeListener.java
----------------------------------------------------------------------
diff --git a/tools/atlas-migration-exporter/src/main/java/org/apache/atlas/migration/NoOpNotificationChangeListener.java b/tools/atlas-migration-exporter/src/main/java/org/apache/atlas/migration/NoOpNotificationChangeListener.java
index 9af7819..782ac4b 100644
--- a/tools/atlas-migration-exporter/src/main/java/org/apache/atlas/migration/NoOpNotificationChangeListener.java
+++ b/tools/atlas-migration-exporter/src/main/java/org/apache/atlas/migration/NoOpNotificationChangeListener.java
@@ -20,6 +20,7 @@ package org.apache.atlas.migration;
import org.apache.atlas.AtlasException;
import org.apache.atlas.listener.EntityChangeListener;
+import org.apache.atlas.model.glossary.AtlasGlossaryTerm;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.v1.model.instance.Struct;
import org.springframework.stereotype.Component;
@@ -57,4 +58,14 @@ public class NoOpNotificationChangeListener implements EntityChangeListener {
public void onEntitiesDeleted(Collection<Referenceable> entities, boolean isImport) throws AtlasException {
}
+
+ @Override
+ public void onTermAdded(Collection<Referenceable> entities, AtlasGlossaryTerm term) throws AtlasException {
+
+ }
+
+ @Override
+ public void onTermDeleted(Collection<Referenceable> entities, AtlasGlossaryTerm term) throws AtlasException {
+
+ }
}
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java
----------------------------------------------------------------------
diff --git a/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java b/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java
index 9de49d4..017de99 100644
--- a/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java
+++ b/webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java
@@ -21,9 +21,11 @@ import org.apache.atlas.AtlasErrorCode;
import org.apache.atlas.RequestContextV1;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.listener.EntityChangeListenerV2;
+import org.apache.atlas.model.glossary.AtlasGlossaryTerm;
import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.model.instance.AtlasEntityHeader;
+import org.apache.atlas.model.instance.AtlasRelatedObjectId;
import org.apache.atlas.model.notification.EntityNotification.EntityNotificationV2;
import org.apache.atlas.model.notification.EntityNotification.EntityNotificationV2.OperationType;
import org.apache.atlas.type.AtlasClassificationType;
@@ -102,6 +104,16 @@ public class EntityNotificationListenerV2 implements EntityChangeListenerV2 {
notifyEntityEvents(Collections.singletonList(entity), CLASSIFICATION_DELETE);
}
+ @Override
+ public void onTermAdded(AtlasGlossaryTerm term, List<AtlasRelatedObjectId> entities) {
+ // do nothing -> notification not sent out for term assignment to entities
+ }
+
+ @Override
+ public void onTermDeleted(AtlasGlossaryTerm term, List<AtlasRelatedObjectId> entities) {
+ // do nothing -> notification not sent out for term removal from entities
+ }
+
private void notifyEntityEvents(List<AtlasEntity> entities, OperationType operationType) throws AtlasBaseException {
List<EntityNotificationV2> messages = new ArrayList<>();
http://git-wip-us.apache.org/repos/asf/atlas/blob/1deb8c16/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
----------------------------------------------------------------------
diff --git a/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java b/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
index dbdf67d..1eeecef 100644
--- a/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
+++ b/webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
@@ -21,6 +21,7 @@ import com.google.common.annotations.VisibleForTesting;
import org.apache.atlas.ApplicationProperties;
import org.apache.atlas.AtlasException;
import org.apache.atlas.listener.EntityChangeListener;
+import org.apache.atlas.model.glossary.AtlasGlossaryTerm;
import org.apache.atlas.notification.NotificationInterface.NotificationType;
import org.apache.atlas.v1.model.instance.Referenceable;
import org.apache.atlas.v1.model.instance.Struct;
@@ -99,6 +100,16 @@ public class NotificationEntityChangeListener implements EntityChangeListener {
notifyOfEntityEvent(entities, OperationType.ENTITY_DELETE);
}
+ @Override
+ public void onTermAdded(Collection<Referenceable> entities, AtlasGlossaryTerm term) throws AtlasException {
+ // do nothing
+ }
+
+ @Override
+ public void onTermDeleted(Collection<Referenceable> entities, AtlasGlossaryTerm term) throws AtlasException {
+ // do nothing
+ }
+
// ----- helper methods -------------------------------------------------