Skip to content

A
atlas
Commit a3374c74 by Sarath Subramanian Committed by Madhan Neethiraj
Options

ATLAS-2456: Implement tag propagation using relationships

Signed-off-by: 's avatarMadhan Neethiraj <madhan@apache.org>
parent 9c58d30c
Showing with 1317 additions and 158 deletions
addons/models/0000-Area0/0010-base_model.json
......@@ -213,7 +213,7 @@
"cardinality": "SET",
"isLegacyAttribute": true
},
"propagateTags": "NONE"
"propagateTags": "ONE_TO_TWO"
},
{
"name": "process_dataset_outputs",
......@@ -232,7 +232,7 @@
"isContainer": false,
"cardinality": "SET"
},
"propagateTags": "NONE"
"propagateTags": "ONE_TO_TWO"
}
]
}
addons/models/1000-Hadoop/1030-hive_model.json
......@@ -539,7 +539,7 @@
"cardinality": "SINGLE",
"isLegacyAttribute": true
},
"propagateTags": "ONE_TO_TWO"
"propagateTags": "NONE"
},
{
"name": "hive_table_columns",
......@@ -559,7 +559,7 @@
"cardinality": "SINGLE",
"isLegacyAttribute": true
},
"propagateTags": "ONE_TO_TWO"
"propagateTags": "NONE"
},
{
"name": "hive_table_partitionkeys",
......@@ -579,7 +579,7 @@
"cardinality": "SINGLE",
"isLegacyAttribute": true
},
"propagateTags": "ONE_TO_TWO"
"propagateTags": "NONE"
},
{
"name": "hive_table_storagedesc",
......@@ -599,7 +599,7 @@
"cardinality": "SINGLE",
"isLegacyAttribute": true
},
"propagateTags": "ONE_TO_TWO"
"propagateTags": "NONE"
},
{
"name": "hive_process_column_lineage",
......
addons/models/1000-Hadoop/1060-hbase_model.json
......@@ -157,7 +157,7 @@
"cardinality": "SINGLE",
"isLegacyAttribute": true
},
"propagateTags": "ONE_TO_TWO"
"propagateTags": "NONE"
},
{
"name": "hbase_column_family_columns",
......@@ -177,7 +177,7 @@
"cardinality": "SINGLE",
"isLegacyAttribute": true
},
"propagateTags": "ONE_TO_TWO"
"propagateTags": "NONE"
}
]
}
\ No newline at end of file
}
common/src/main/java/org/apache/atlas/repository/Constants.java
......@@ -69,6 +69,7 @@ public final class Constants {
* Trait names property key and index name.
*/
public static final String TRAIT_NAMES_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "traitNames";
public static final String PROPAGATED_TRAIT_NAMES_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "propagatedTraitNames";
public static final String VERSION_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "version";
public static final String STATE_PROPERTY_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "state";
......@@ -115,6 +116,9 @@ public final class Constants {
public static final String ATTRIBUTE_NAME_VERSION = "version";
public static final String TEMP_STRUCT_NAME_PREFIX = "__tempQueryResultStruct";
public static final String CLASSIFICATION_ENTITY_GUID = INTERNAL_PROPERTY_KEY_PREFIX + "entityGuid";
public static final String CLASSIFICATION_PROPAGATE_KEY = INTERNAL_PROPERTY_KEY_PREFIX + "propagate";
private Constants() {
}
......
graphdb/api/src/main/java/org/apache/atlas/repository/graphdb/AtlasElement.java
......@@ -22,7 +22,6 @@ import java.util.Collection;
import java.util.List;
import java.util.Set;
import org.apache.atlas.AtlasException;
import org.codehaus.jettison.json.JSONException;
import org.codehaus.jettison.json.JSONObject;
......@@ -85,7 +84,7 @@ public interface AtlasElement {
* is needed for this because special logic is required to handle this situation
* in some implementations.
*/
void setListProperty(String propertyName, List<String> values) throws AtlasException;
void setListProperty(String propertyName, List<String> values);
/**
......
graphdb/api/src/main/java/org/apache/atlas/repository/graphdb/AtlasVertex.java
......@@ -55,6 +55,14 @@ public interface AtlasVertex<V, E> extends AtlasElement {
*/
<T> void addProperty(String propertyName, T value);
/**
* Adds a value to a multiplicity-many property.
* If the property is already present, the value is added to it; if not, the propery is set with the given value
*
* @param propertyName
* @param value
*/
<T> void addListProperty(String propertyName, T value);
/**
* Creates a vertex query.
......
graphdb/janus/src/main/java/org/apache/atlas/repository/graphdb/janus/AtlasJanusVertex.java
......@@ -52,6 +52,14 @@ public class AtlasJanusVertex extends AtlasJanusElement<Vertex> implements Atlas
}
}
@Override
public <T> void addListProperty(String propertyName, T value) {
try {
getWrappedElement().property(VertexProperty.Cardinality.list, propertyName, value);
} catch(SchemaViolationException e) {
throw new AtlasSchemaViolationException(e);
}
}
@Override
......
graphdb/titan0/src/main/java/org/apache/atlas/repository/graphdb/titan0/Titan0Vertex.java
......@@ -100,6 +100,20 @@ public class Titan0Vertex extends Titan0Element<Vertex> implements AtlasVertex<T
}
@Override
public <T> void addListProperty(String propertyName, T value) {
try {
getAsTitanVertex().addProperty(propertyName, value);
} catch (SchemaViolationException e) {
if (getPropertyValues(propertyName, value.getClass()).contains(value)) {
// follow java set semantics, don't throw an exception if
// value is already there.
return;
}
throw new AtlasSchemaViolationException(e);
}
}
@Override
public <T> Collection<T> getPropertyValues(String key, Class<T> clazz) {
TitanVertex tv = getAsTitanVertex();
......
intg/src/main/java/org/apache/atlas/AtlasErrorCode.java
......@@ -122,6 +122,10 @@ public enum AtlasErrorCode {
INVALID_DSL_HAS_PROPERTY(400, "ATLAS-400-00-068", "DSL Semantic Error - Property needs to be a primitive type: {0}"),
RELATIONSHIP_UPDATE_END_CHANGE_NOT_ALLOWED(404, "ATLAS-400-00-069", "change of relationship end is not permitted. relationship-type={}, relationship-guid={}, end-guid={}, updated-end-guid={}"),
RELATIONSHIP_UPDATE_TYPE_CHANGE_NOT_ALLOWED(404, "ATLAS-400-00-06A", "change of relationship type is not permitted. relationship-guid={}, current-type={}, new-type={}"),
CLASSIFICATION_UPDATE_FROM_PROPAGATED_ENTITY(400, "ATLAS-400-00-06B", "Update to classification {0} is not allowed from propagated entity"),
CLASSIFICATION_DELETE_FROM_PROPAGATED_ENTITY(400, "ATLAS-400-00-06C", "Delete of classification {0} is not allowed from propagated entity"),
CLASSIFICATION_NOT_ASSOCIATED_WITH_ENTITY(400, "ATLAS-400-00-06D", "Classification {0} is not associated with entity"),
// All Not found enums go here
TYPE_NAME_NOT_FOUND(404, "ATLAS-404-00-001", "Given typename {0} was invalid"),
TYPE_GUID_NOT_FOUND(404, "ATLAS-404-00-002", "Given type guid {0} was invalid"),
......@@ -137,6 +141,7 @@ public enum AtlasErrorCode {
RELATIONSHIP_CRUD_INVALID_PARAMS(404, "ATLAS-404-00-00D", "Invalid relationship creation/updation parameters passed : {0}"),
RELATIONSHIPDEF_END_TYPE_NAME_NOT_FOUND(404, "ATLAS-404-00-00E", "RelationshipDef {0} endDef typename {0} cannot be found"),
RELATIONSHIP_ALREADY_DELETED(404, "ATLAS-404-00-00F", "Attempting to delete a relationship which is already deleted : {0}"),
INVALID_ENTITY_GUID_FOR_CLASSIFICATION_UPDATE(404, "ATLAS-404-00-010", "Updating entityGuid of classification is not allowed."),
// All data conflict errors go here
TYPE_ALREADY_EXISTS(409, "ATLAS-409-00-001", "Given type {0} already exists"),
......@@ -157,7 +162,6 @@ public enum AtlasErrorCode {
FAILED_TO_OBTAIN_GREMLIN_SCRIPT_ENGINE(500, "ATLAS-500-00-008", "Failed to obtain gremlin script engine: {0}"),
JSON_ERROR_OBJECT_MAPPER_NULL_RETURNED(500, "ATLAS-500-00-009", "ObjectMapper.readValue returned NULL for class: {0}"),
GREMLIN_SCRIPT_EXECUTION_FAILED(500, "ATLAS-500-00-00A", "Gremlin script execution failed: {0}"),
CURATOR_FRAMEWORK_UPDATE(500, "ATLAS-500-00-00B", "ActiveInstanceState.update resulted in exception."),
QUICK_START(500, "ATLAS-500-00-00C", "Failed to run QuickStart: {0}"),
EMBEDDED_SERVER_START(500, "ATLAS-500-00-00D", "EmbeddedServer.Start: failed!"),
......@@ -165,9 +169,9 @@ public enum AtlasErrorCode {
SQOOP_HOOK(500, "ATLAS-500-00-00F", "SqoopHook: {0}"),
HIVE_HOOK(500, "ATLAS-500-00-010", "HiveHook: {0}"),
HIVE_HOOK_METASTORE_BRIDGE(500, "ATLAS-500-00-011", "HiveHookMetaStoreBridge: {0}"),
DATA_ACCESS_SAVE_FAILED(500, "ATLAS-500-00-00B", "Save failed: {0}"),
DATA_ACCESS_LOAD_FAILED(500, "ATLAS-500-00-00C", "Load failed: {0}");
DATA_ACCESS_SAVE_FAILED(500, "ATLAS-500-00-012", "Save failed: {0}"),
DATA_ACCESS_LOAD_FAILED(500, "ATLAS-500-00-013", "Load failed: {0}"),
ENTITY_NOTIFICATION_FAILED(500, "ATLAS-500-00-014", "Notification failed for operation: {} : {}");
private String errorCode;
private String errorMessage;
......
intg/src/main/java/org/apache/atlas/listener/EntityChangeListenerV2.java 0 → 100644
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
* <p>
* http://www.apache.org/licenses/LICENSE-2.0
* <p>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.atlas.listener;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntity;
import java.util.List;
/**
* Entity change notification listener V2.
*/
public interface EntityChangeListenerV2 {
/**
* This is upon adding new entities to the repository.
*
* @param entities the created entities
* @param isImport
*/
void onEntitiesAdded(List<AtlasEntity> entities, boolean isImport) throws AtlasBaseException;
/**
* This is upon updating an entity.
*
* @param entities the updated entities
* @param isImport
*/
void onEntitiesUpdated(List<AtlasEntity> entities, boolean isImport) throws AtlasBaseException;
/**
* This is upon deleting entities from the repository.
*
* @param entities the deleted entities
* @param isImport
*/
void onEntitiesDeleted(List<AtlasEntity> entities, boolean isImport) throws AtlasBaseException;
/**
* This is upon adding new classifications to an entity.
*
* @param entity the entity
* @param classifications classifications that needs to be added to an entity
* @throws AtlasBaseException if the listener notification fails
*/
void onClassificationsAdded(AtlasEntity entity, List<AtlasClassification> classifications) throws AtlasBaseException;
/**
* This is upon updating classifications to an entity.
*
* @param entity the entity
* @param classifications classifications that needs to be updated for an entity
* @throws AtlasBaseException if the listener notification fails
*/
void onClassificationsUpdated(AtlasEntity entity, List<AtlasClassification> classifications) throws AtlasBaseException;
/**
* This is upon deleting classifications from an entity.
*
* @param entity the entity
* @param classificationNames classifications names for the instance that needs to be deleted from entity
* @throws AtlasBaseException if the listener notification fails
*/
void onClassificationsDeleted(AtlasEntity entity, List<String> classificationNames) throws AtlasBaseException;
}
\ No newline at end of file
intg/src/main/java/org/apache/atlas/model/audit/EntityAuditEventV2.java 0 → 100644
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
* <p/>
* http://www.apache.org/licenses/LICENSE-2.0
* <p/>
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.atlas.model.audit;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.type.AtlasType;
import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlRootElement;
import java.io.Serializable;
import java.util.Objects;
import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.NONE;
import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.PUBLIC_ONLY;
/**
* Structure of v2 entity audit event
*/
@JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
@JsonSerialize(include=JsonSerialize.Inclusion.ALWAYS)
@JsonIgnoreProperties(ignoreUnknown=true)
@XmlRootElement
@XmlAccessorType(XmlAccessType.PROPERTY)
public class EntityAuditEventV2 implements Serializable {
public enum EntityAuditAction {
ENTITY_CREATE, ENTITY_UPDATE, ENTITY_DELETE,
ENTITY_IMPORT_CREATE, ENTITY_IMPORT_UPDATE, ENTITY_IMPORT_DELETE,
CLASSIFICATION_ADD, CLASSIFICATION_DELETE, CLASSIFICATION_UPDATE,
}
private String entityId;
private long timestamp;
private String user;
private EntityAuditAction action;
private String details;
private String eventKey;
private AtlasEntity entity;
public EntityAuditEventV2() { }
public EntityAuditEventV2(String entityId, long timestamp, String user, EntityAuditAction action, String details,
AtlasEntity entity) {
setEntityId(entityId);
setTimestamp(timestamp);
setUser(user);
setAction(action);
setDetails(details);
setEntity(entity);
}
public String getEntityId() {
return entityId;
}
public void setEntityId(String entityId) {
this.entityId = entityId;
}
public long getTimestamp() {
return timestamp;
}
public void setTimestamp(long timestamp) {
this.timestamp = timestamp;
}
public String getUser() {
return user;
}
public void setUser(String user) {
this.user = user;
}
public EntityAuditAction getAction() {
return action;
}
public void setAction(EntityAuditAction action) {
this.action = action;
}
public String getDetails() {
return details;
}
public void setDetails(String details) {
this.details = details;
}
public String getEventKey() {
return eventKey;
}
public void setEventKey(String eventKey) {
this.eventKey = eventKey;
}
public AtlasEntity getEntity() {
return entity;
}
public void setEntity(AtlasEntity entity) {
this.entity = entity;
}
@JsonIgnore
public String getEntityDefinitionString() {
if (entity != null) {
return AtlasType.toJson(entity);
}
return null;
}
@JsonIgnore
public void setEntityDefinition(String entityDefinition) {
this.entity = AtlasType.fromJson(entityDefinition, AtlasEntity.class);
}
@Override
public boolean equals(Object o) {
if (this == o) { return true; }
if (o == null || getClass() != o.getClass()) { return false; }
EntityAuditEventV2 that = (EntityAuditEventV2) o;
return timestamp == that.timestamp &&
Objects.equals(entityId, that.entityId) &&
Objects.equals(user, that.user) &&
action == that.action &&
Objects.equals(details, that.details) &&
Objects.equals(eventKey, that.eventKey) &&
Objects.equals(entity, that.entity);
}
@Override
public int hashCode() {
return Objects.hash(entityId, timestamp, user, action, details, eventKey, entity);
}
@Override
public String toString() {
final StringBuilder sb = new StringBuilder("EntityAuditEventV2{");
sb.append("entityId='").append(entityId).append('\'');
sb.append(", timestamp=").append(timestamp);
sb.append(", user='").append(user).append('\'');
sb.append(", action=").append(action);
sb.append(", details='").append(details).append('\'');
sb.append(", eventKey='").append(eventKey).append('\'');
sb.append(", entity=").append(entity);
sb.append('}');
return sb.toString();
}
}
\ No newline at end of file
intg/src/main/java/org/apache/atlas/model/instance/AtlasClassification.java
......@@ -25,6 +25,7 @@ import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
......@@ -49,6 +50,9 @@ import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.PUBLIC_
public class AtlasClassification extends AtlasStruct implements Serializable {
private static final long serialVersionUID = 1L;
private String entityGuid = null;
private boolean propagate = true;
public AtlasClassification() {
this(null, null);
}
......@@ -76,6 +80,47 @@ public class AtlasClassification extends AtlasStruct implements Serializable {
}
}
public String getEntityGuid() {
return entityGuid;
}
public void setEntityGuid(String entityGuid) {
this.entityGuid = entityGuid;
}
public boolean isPropagate() {
return propagate;
}
public void setPropagate(boolean propagate) {
this.propagate = propagate;
}
@Override
public boolean equals(Object o) {
if (this == o) { return true; }
if (o == null || getClass() != o.getClass()) { return false; }
if (!super.equals(o)) { return false; }
AtlasClassification that = (AtlasClassification) o;
return propagate == that.propagate &&
Objects.equals(entityGuid, that.entityGuid);
}
@Override
public int hashCode() {
return Objects.hash(super.hashCode(), entityGuid, propagate);
}
@Override
public String toString() {
final StringBuilder sb = new StringBuilder("AtlasClassification{");
super.toString(sb);
sb.append("entityGuid='").append(entityGuid).append('\'');
sb.append(", propagate=").append(propagate);
sb.append('}');
return sb.toString();
}
/**
* REST serialization friendly list.
*/
......
intg/src/main/java/org/apache/atlas/model/instance/AtlasEntity.java
......@@ -259,6 +259,17 @@ public class AtlasEntity extends AtlasStruct implements Serializable {
public void setClassifications(List<AtlasClassification> classifications) { this.classifications = classifications; }
public void addClassifications(List<AtlasClassification> classifications) {
List<AtlasClassification> c = this.classifications;
if (c == null) {
c = new ArrayList<>(classifications);
this.classifications = c;
}
c.addAll(classifications);
}
private void init() {
setGuid(nextInternalId());
......
intg/src/main/java/org/apache/atlas/model/notification/EntityNotification.java
......@@ -44,7 +44,7 @@ public class EntityNotification implements Serializable {
* Type of the hook message.
*/
public enum EntityNotificationType {
ENTITY_NOTIFICATION_V1
ENTITY_NOTIFICATION_V1, ENTITY_NOTIFICATION_V2
}
protected EntityNotificationType type;
......
intg/src/main/java/org/apache/atlas/v1/model/notification/EntityNotificationV2.java 0 → 100644
/**
* 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.atlas.v1.model.notification;
import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.model.notification.EntityNotification;
import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
import javax.xml.bind.annotation.XmlAccessType;
import javax.xml.bind.annotation.XmlAccessorType;
import javax.xml.bind.annotation.XmlRootElement;
import java.io.Serializable;
import java.util.List;
import java.util.Objects;
import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.NONE;
import static com.fasterxml.jackson.annotation.JsonAutoDetect.Visibility.PUBLIC_ONLY;
import static org.apache.atlas.model.notification.EntityNotification.EntityNotificationType.ENTITY_NOTIFICATION_V2;
/**
* Entity v2 notification
*/
@JsonAutoDetect(getterVisibility=PUBLIC_ONLY, setterVisibility=PUBLIC_ONLY, fieldVisibility=NONE)
@JsonSerialize(include=JsonSerialize.Inclusion.NON_NULL)
@JsonIgnoreProperties(ignoreUnknown=true)
@XmlRootElement
@XmlAccessorType(XmlAccessType.PROPERTY)
public class EntityNotificationV2 extends EntityNotification implements Serializable {
private static final long serialVersionUID = 1L;
public enum OperationType {
ENTITY_CREATE, ENTITY_UPDATE, ENTITY_DELETE,
CLASSIFICATION_ADD, CLASSIFICATION_DELETE, CLASSIFICATION_UPDATE
}
private AtlasEntity entity;
private OperationType operationType;
private List<AtlasClassification> classifications;
public EntityNotificationV2() { }
public EntityNotificationV2(AtlasEntity entity, OperationType operationType, List<AtlasClassification> classifications) {
setEntity(entity);
setOperationType(operationType);
setClassifications(classifications);
setType(ENTITY_NOTIFICATION_V2);
}
public AtlasEntity getEntity() {
return entity;
}
public void setEntity(AtlasEntity entity) {
this.entity = entity;
}
public OperationType getOperationType() {
return operationType;
}
public void setOperationType(OperationType operationType) {
this.operationType = operationType;
}
public List<AtlasClassification> getClassifications() {
return classifications;
}
public void setClassifications(List<AtlasClassification> classifications) {
this.classifications = classifications;
}
@Override
public boolean equals(Object o) {
if (this == o) { return true; }
if (o == null || getClass() != o.getClass()) { return false; }
EntityNotificationV2 that = (EntityNotificationV2) o;
return Objects.equals(entity, that.entity) &&
operationType == that.operationType &&
Objects.equals(classifications, that.classifications);
}
@Override
public int hashCode() {
return Objects.hash(entity, operationType, classifications);
}
@Override
public StringBuilder toString(StringBuilder sb) {
if (sb == null) {
sb = new StringBuilder();
}
sb.append("EntityNotificationV1{");
super.toString(sb);
sb.append(", entity=");
if (entity != null) {
entity.toString(sb);
} else {
sb.append(entity);
}
sb.append(", operationType=").append(operationType);
sb.append(", classifications=[");
AtlasBaseTypeDef.dumpObjects(classifications, sb);
sb.append("]");
sb.append("}");
return sb;
}
}
\ No newline at end of file
repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListener.java
......@@ -67,7 +67,7 @@ public class EntityAuditListener implements EntityChangeListener {
events.add(event);
}
auditRepository.putEvents(events);
auditRepository.putEventsV1(events);
}
@Override
......@@ -78,7 +78,7 @@ public class EntityAuditListener implements EntityChangeListener {
events.add(event);
}
auditRepository.putEvents(events);
auditRepository.putEventsV1(events);
}
@Override
......@@ -88,7 +88,7 @@ public class EntityAuditListener implements EntityChangeListener {
EntityAuditEvent event = createEvent(entity, EntityAuditAction.TAG_ADD,
"Added trait: " + AtlasType.toV1Json(trait));
auditRepository.putEvents(event);
auditRepository.putEventsV1(event);
}
}
}
......@@ -99,7 +99,7 @@ public class EntityAuditListener implements EntityChangeListener {
for (String traitName : traitNames) {
EntityAuditEvent event = createEvent(entity, EntityAuditAction.TAG_DELETE, "Deleted trait: " + traitName);
auditRepository.putEvents(event);
auditRepository.putEventsV1(event);
}
}
}
......@@ -111,7 +111,7 @@ public class EntityAuditListener implements EntityChangeListener {
EntityAuditEvent event = createEvent(entity, EntityAuditAction.TAG_UPDATE,
"Updated trait: " + AtlasType.toV1Json(trait));
auditRepository.putEvents(event);
auditRepository.putEventsV1(event);
}
}
}
......@@ -124,11 +124,11 @@ public class EntityAuditListener implements EntityChangeListener {
events.add(event);
}
auditRepository.putEvents(events);
auditRepository.putEventsV1(events);
}
public List<EntityAuditEvent> getAuditEvents(String guid) throws AtlasException{
return auditRepository.listEvents(guid, null, (short) 10);
return auditRepository.listEventsV1(guid, null, (short) 10);
}
private EntityAuditEvent createEvent(Referenceable entity, EntityAuditAction action)
......
repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditRepository.java
......@@ -20,6 +20,8 @@ package org.apache.atlas.repository.audit;
import org.apache.atlas.AtlasException;
import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.model.audit.EntityAuditEventV2;
import org.apache.atlas.exception.AtlasBaseException;
import java.util.List;
......@@ -32,14 +34,14 @@ public interface EntityAuditRepository {
* @param events events to be added
* @throws AtlasException
*/
void putEvents(EntityAuditEvent... events) throws AtlasException;
void putEventsV1(EntityAuditEvent... events) throws AtlasException;
/**
* Add events to the event repository
* @param events events to be added
* @throws AtlasException
*/
void putEvents(List<EntityAuditEvent> events) throws AtlasException;
void putEventsV1(List<EntityAuditEvent> events) throws AtlasException;
/**
* List events for the given entity id in decreasing order of timestamp, from the given timestamp. Returns n results
......@@ -49,13 +51,48 @@ public interface EntityAuditRepository {
* @return list of events
* @throws AtlasException
*/
List<EntityAuditEvent> listEvents(String entityId, String startKey, short n) throws AtlasException;
List<EntityAuditEvent> listEventsV1(String entityId, String startKey, short n) throws AtlasException;
/**
* Add v2 events to the event repository
* @param events events to be added
* @throws AtlasBaseException
*/
void putEventsV2(EntityAuditEventV2... events) throws AtlasBaseException;
/**
* Add v2 events to the event repository
* @param events events to be added
* @throws AtlasBaseException
*/
void putEventsV2(List<EntityAuditEventV2> events) throws AtlasBaseException;
/**
* List events for the given entity id in decreasing order of timestamp, from the given timestamp. Returns n results
* @param entityId entity id
* @param startKey key for the first event to be returned, used for pagination
* @param n number of events to be returned
* @return list of events
* @throws AtlasBaseException
*/
List<EntityAuditEventV2> listEventsV2(String entityId, String startKey, short n) throws AtlasBaseException;
/**
* List events for the given entity id in decreasing order of timestamp, from the given timestamp. Returns n results
* @param entityId entity id
* @param startKey key for the first event to be returned, used for pagination
* @param n number of events to be returned
* @return list of events
* @throws AtlasBaseException
*/
List<Object> listEvents(String entityId, String startKey, short n) throws AtlasBaseException;
/**
* Returns maximum allowed repository size per EntityAuditEvent
* @throws AtlasException
*/
long repositoryMaxSize() throws AtlasException;
long repositoryMaxSize();
/**
* list of attributes to be excluded when storing in audit repo.
......@@ -63,5 +100,5 @@ public interface EntityAuditRepository {
* @return list of attribute names to be excluded
* @throws AtlasException
*/
List<String> getAuditExcludeAttributes(String entityType) throws AtlasException;
List<String> getAuditExcludeAttributes(String entityType);
}
repository/src/main/java/org/apache/atlas/repository/audit/HBaseBasedAuditRepository.java
......@@ -23,9 +23,13 @@ import org.apache.atlas.ApplicationProperties;
import org.apache.atlas.AtlasException;
import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.annotation.ConditionalOnAtlasProperty;
import org.apache.atlas.model.audit.EntityAuditEventV2;
import org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditAction;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.ha.HAConfiguration;
import org.apache.atlas.listener.ActiveStateChangeHandler;
import org.apache.atlas.service.Service;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.HBaseConfiguration;
......@@ -116,17 +120,17 @@ public class HBaseBasedAuditRepository implements Service, EntityAuditRepository
* @throws AtlasException
*/
@Override
public void putEvents(EntityAuditEvent... events) throws AtlasException {
putEvents(Arrays.asList(events));
public void putEventsV1(EntityAuditEvent... events) throws AtlasException {
putEventsV1(Arrays.asList(events));
}
@Override
/**
* Add events to the event repository
* @param events events to be added
* @throws AtlasException
*/
public void putEvents(List<EntityAuditEvent> events) throws AtlasException {
@Override
public void putEventsV1(List<EntityAuditEvent> events) throws AtlasException {
if (LOG.isDebugEnabled()) {
LOG.debug("Putting {} events", events.size());
}
......@@ -154,6 +158,146 @@ public class HBaseBasedAuditRepository implements Service, EntityAuditRepository
}
}
@Override
public void putEventsV2(EntityAuditEventV2... events) throws AtlasBaseException {
putEventsV2(Arrays.asList(events));
}
@Override
public void putEventsV2(List<EntityAuditEventV2> events) throws AtlasBaseException {
if (LOG.isDebugEnabled()) {
LOG.debug("Putting {} events", events.size());
}
Table table = null;
try {
table = connection.getTable(tableName);
List<Put> puts = new ArrayList<>(events.size());
for (EntityAuditEventV2 event : events) {
if (LOG.isDebugEnabled()) {
LOG.debug("Adding entity audit event {}", event);
}
Put put = new Put(getKey(event.getEntityId(), event.getTimestamp()));
addColumn(put, COLUMN_ACTION, event.getAction());
addColumn(put, COLUMN_USER, event.getUser());
addColumn(put, COLUMN_DETAIL, event.getDetails());
if (persistEntityDefinition) {
addColumn(put, COLUMN_DEFINITION, event.getEntity());
}
puts.add(put);
}
table.put(puts);
} catch (IOException e) {
throw new AtlasBaseException(e);
} finally {
try {
close(table);
} catch (AtlasException e) {
throw new AtlasBaseException(e);
}
}
}
@Override
public List<EntityAuditEventV2> listEventsV2(String entityId, String startKey, short n) throws AtlasBaseException {
if (LOG.isDebugEnabled()) {
LOG.debug("Listing events for entity id {}, starting timestamp {}, #records {}", entityId, startKey, n);
}
Table table = null;
ResultScanner scanner = null;
try {
table = connection.getTable(tableName);
/**
* Scan Details:
* In hbase, the events are stored in increasing order of timestamp. So, doing reverse scan to get the latest event first
* Page filter is set to limit the number of results returned.
* Stop row is set to the entity id to avoid going past the current entity while scanning
* small is set to true to optimise RPC calls as the scanner is created per request
*/
Scan scan = new Scan().setReversed(true).setFilter(new PageFilter(n))
.setStopRow(Bytes.toBytes(entityId))
.setCaching(n)
.setSmall(true);
if (StringUtils.isEmpty(startKey)) {
//Set start row to entity id + max long value
byte[] entityBytes = getKey(entityId, Long.MAX_VALUE);
scan = scan.setStartRow(entityBytes);
} else {
scan = scan.setStartRow(Bytes.toBytes(startKey));
}
scanner = table.getScanner(scan);
List<EntityAuditEventV2> events = new ArrayList<>();
Result result;
//PageFilter doesn't ensure n results are returned. The filter is per region server.
//So, adding extra check on n here
while ((result = scanner.next()) != null && events.size() < n) {
EntityAuditEventV2 event = fromKeyV2(result.getRow());
//In case the user sets random start key, guarding against random events
if (!event.getEntityId().equals(entityId)) {
continue;
}
event.setUser(getResultString(result, COLUMN_USER));
event.setAction(EntityAuditAction.valueOf(getResultString(result, COLUMN_ACTION)));
event.setDetails(getResultString(result, COLUMN_DETAIL));
if (persistEntityDefinition) {
String colDef = getResultString(result, COLUMN_DEFINITION);
if (colDef != null) {
event.setEntityDefinition(colDef);
}
}
events.add(event);
}
if (LOG.isDebugEnabled()) {
LOG.debug("Got events for entity id {}, starting timestamp {}, #records {}", entityId, startKey, events.size());
}
return events;
} catch (IOException e) {
throw new AtlasBaseException(e);
} finally {
try {
close(scanner);
close(table);
} catch (AtlasException e) {
throw new AtlasBaseException(e);
}
}
}
@Override
public List<Object> listEvents(String entityId, String startKey, short maxResults) throws AtlasBaseException {
List ret = listEventsV2(entityId, startKey, maxResults);
try {
if (CollectionUtils.isEmpty(ret)) {
ret = listEventsV1(entityId, startKey, maxResults);
}
} catch (AtlasException e) {
throw new AtlasBaseException(e);
}
return ret;
}
private <T> void addColumn(Put put, byte[] columnName, T columnValue) {
if (columnValue != null && !columnValue.toString().isEmpty()) {
put.addColumn(COLUMN_FAMILY, columnName, Bytes.toBytes(columnValue.toString()));
......@@ -175,7 +319,7 @@ public class HBaseBasedAuditRepository implements Service, EntityAuditRepository
* @return list of events
* @throws AtlasException
*/
public List<EntityAuditEvent> listEvents(String entityId, String startKey, short n)
public List<EntityAuditEvent> listEventsV1(String entityId, String startKey, short n)
throws AtlasException {
if (LOG.isDebugEnabled()) {
LOG.debug("Listing events for entity id {}, starting timestamp {}, #records {}", entityId, startKey, n);
......@@ -243,7 +387,7 @@ public class HBaseBasedAuditRepository implements Service, EntityAuditRepository
}
@Override
public long repositoryMaxSize() throws AtlasException {
public long repositoryMaxSize() {
long ret;
initApplicationProperties();
......@@ -257,7 +401,7 @@ public class HBaseBasedAuditRepository implements Service, EntityAuditRepository
}
@Override
public List<String> getAuditExcludeAttributes(String entityType) throws AtlasException {
public List<String> getAuditExcludeAttributes(String entityType) {
List<String> ret = null;
initApplicationProperties();
......@@ -308,6 +452,20 @@ public class HBaseBasedAuditRepository implements Service, EntityAuditRepository
return event;
}
private EntityAuditEventV2 fromKeyV2(byte[] keyBytes) {
String key = Bytes.toString(keyBytes);
EntityAuditEventV2 event = new EntityAuditEventV2();
if (StringUtils.isNotEmpty(key)) {
String[] parts = key.split(FIELD_SEPARATOR);
event.setEntityId(parts[0]);
event.setTimestamp(Long.valueOf(parts[1]));
event.setEventKey(key);
}
return event;
}
private void close(Closeable closeable) throws AtlasException {
if (closeable != null) {
try {
......
repository/src/main/java/org/apache/atlas/repository/audit/InMemoryEntityAuditRepository.java
......@@ -21,6 +21,8 @@ package org.apache.atlas.repository.audit;
import org.apache.atlas.AtlasException;
import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.annotation.ConditionalOnAtlasProperty;
import org.apache.atlas.model.audit.EntityAuditEventV2;
import org.apache.commons.collections.CollectionUtils;
import org.springframework.stereotype.Component;
import javax.inject.Singleton;
......@@ -37,15 +39,16 @@ import java.util.TreeMap;
@Component
@ConditionalOnAtlasProperty(property = "atlas.EntityAuditRepository.impl")
public class InMemoryEntityAuditRepository implements EntityAuditRepository {
private TreeMap<String, EntityAuditEvent> auditEvents = new TreeMap<>();
private TreeMap<String, EntityAuditEvent> auditEvents = new TreeMap<>();
private TreeMap<String, EntityAuditEventV2> auditEventsV2 = new TreeMap<>();
@Override
public void putEvents(EntityAuditEvent... events) throws AtlasException {
putEvents(Arrays.asList(events));
public void putEventsV1(EntityAuditEvent... events) throws AtlasException {
putEventsV1(Arrays.asList(events));
}
@Override
public synchronized void putEvents(List<EntityAuditEvent> events) throws AtlasException {
public synchronized void putEventsV1(List<EntityAuditEvent> events) throws AtlasException {
for (EntityAuditEvent event : events) {
String rowKey = event.getEntityId() + (Long.MAX_VALUE - event.getTimestamp());
event.setEventKey(rowKey);
......@@ -56,8 +59,7 @@ public class InMemoryEntityAuditRepository implements EntityAuditRepository {
//synchronized to avoid concurrent modification exception that occurs if events are added
//while we are iterating through the map
@Override
public synchronized List<EntityAuditEvent> listEvents(String entityId, String startKey, short maxResults)
throws AtlasException {
public synchronized List<EntityAuditEvent> listEventsV1(String entityId, String startKey, short maxResults) {
List<EntityAuditEvent> events = new ArrayList<>();
String myStartKey = startKey;
if (myStartKey == null) {
......@@ -73,12 +75,57 @@ public class InMemoryEntityAuditRepository implements EntityAuditRepository {
}
@Override
public long repositoryMaxSize() throws AtlasException {
public long repositoryMaxSize() {
return -1;
}
@Override
public List<String> getAuditExcludeAttributes(String entityType) throws AtlasException {
public List<String> getAuditExcludeAttributes(String entityType) {
return null;
}
@Override
public void putEventsV2(EntityAuditEventV2... events) {
putEventsV2(Arrays.asList(events));
}
@Override
public void putEventsV2(List<EntityAuditEventV2> events) {
for (EntityAuditEventV2 event : events) {
String rowKey = event.getEntityId() + (Long.MAX_VALUE - event.getTimestamp());
event.setEventKey(rowKey);
auditEventsV2.put(rowKey, event);
}
}
@Override
public List<EntityAuditEventV2> listEventsV2(String entityId, String startKey, short maxResults) {
List<EntityAuditEventV2> events = new ArrayList<>();
String myStartKey = startKey;
if (myStartKey == null) {
myStartKey = entityId;
}
SortedMap<String, EntityAuditEventV2> subMap = auditEventsV2.tailMap(myStartKey);
for (EntityAuditEventV2 event : subMap.values()) {
if (events.size() < maxResults && event.getEntityId().equals(entityId)) {
events.add(event);
}
}
return events;
}
@Override
public List<Object> listEvents(String entityId, String startKey, short maxResults) {
List events = listEventsV2(entityId, startKey, maxResults);
if (CollectionUtils.isEmpty(events)) {
events = listEventsV1(entityId, startKey, maxResults);
}
return events;
}
}
repository/src/main/java/org/apache/atlas/repository/audit/NoopEntityAuditRepository.java
......@@ -18,9 +18,9 @@
package org.apache.atlas.repository.audit;
import org.apache.atlas.AtlasException;
import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.annotation.ConditionalOnAtlasProperty;
import org.apache.atlas.model.audit.EntityAuditEventV2;
import org.springframework.stereotype.Component;
import javax.inject.Singleton;
......@@ -36,28 +36,47 @@ import java.util.List;
public class NoopEntityAuditRepository implements EntityAuditRepository {
@Override
public void putEvents(EntityAuditEvent... events) throws AtlasException {
public void putEventsV1(EntityAuditEvent... events) {
//do nothing
}
@Override
public synchronized void putEvents(List<EntityAuditEvent> events) throws AtlasException {
public synchronized void putEventsV1(List<EntityAuditEvent> events) {
//do nothing
}
@Override
public List<EntityAuditEvent> listEvents(String entityId, String startKey, short maxResults)
throws AtlasException {
public List<EntityAuditEvent> listEventsV1(String entityId, String startKey, short maxResults) {
return Collections.emptyList();
}
@Override
public long repositoryMaxSize() throws AtlasException {
public void putEventsV2(EntityAuditEventV2... events) {
//do nothing
}
@Override
public void putEventsV2(List<EntityAuditEventV2> events) {
//do nothing
}
@Override
public List<EntityAuditEventV2> listEventsV2(String entityId, String startKey, short n) {
return Collections.emptyList();
}
@Override
public List<Object> listEvents(String entityId, String startKey, short n) {
return Collections.emptyList();
}
@Override
public long repositoryMaxSize() {
return -1;
}
@Override
public List<String> getAuditExcludeAttributes(String entityType) throws AtlasException {
public List<String> getAuditExcludeAttributes(String entityType) {
return null;
}
}
}
\ No newline at end of file
repository/src/main/java/org/apache/atlas/repository/converters/AtlasInstanceConverter.java
......@@ -20,7 +20,9 @@ package org.apache.atlas.repository.converters;
import org.apache.atlas.AtlasErrorCode;
import org.apache.atlas.AtlasException;
import org.apache.atlas.CreateUpdateEntitiesResult;
import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.RequestContextV1;
import org.apache.atlas.model.audit.EntityAuditEventV2;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.TypeCategory;
import org.apache.atlas.model.instance.AtlasClassification;
......@@ -54,6 +56,13 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
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.model.audit.EntityAuditEventV2.EntityAuditAction.CLASSIFICATION_DELETE;
import static org.apache.atlas.model.audit.EntityAuditEventV2.EntityAuditAction.CLASSIFICATION_UPDATE;
import static org.apache.atlas.v1.model.notification.EntityNotificationV2.OperationType.CLASSIFICATION_ADD;
@Singleton
@Component
public class AtlasInstanceConverter {
......@@ -290,7 +299,7 @@ public class AtlasInstanceConverter {
}
private AtlasEntity.AtlasEntityWithExtInfo getAndCacheEntity(String guid) throws AtlasBaseException {
public AtlasEntity.AtlasEntityWithExtInfo getAndCacheEntity(String guid) throws AtlasBaseException {
RequestContextV1 context = RequestContextV1.get();
AtlasEntity.AtlasEntityWithExtInfo entityWithExtInfo = context.getInstanceV2(guid);
......@@ -308,4 +317,93 @@ public class AtlasInstanceConverter {
return entityWithExtInfo;
}
}
public EntityAuditEvent toV1AuditEvent(EntityAuditEventV2 v2Event) throws AtlasBaseException {
EntityAuditEvent ret = new EntityAuditEvent();
ret.setEntityId(v2Event.getEntityId());
ret.setTimestamp(v2Event.getTimestamp());
ret.setUser(v2Event.getUser());
ret.setDetails(v2Event.getDetails());
ret.setEventKey(v2Event.getEventKey());
ret.setAction(getV1AuditAction(v2Event.getAction()));
ret.setEntityDefinition(getReferenceable(v2Event.getEntityId()));
return ret;
}
public EntityAuditEventV2 toV2AuditEvent(EntityAuditEvent v1Event) throws AtlasBaseException {
EntityAuditEventV2 ret = new EntityAuditEventV2();
ret.setEntityId(v1Event.getEntityId());
ret.setTimestamp(v1Event.getTimestamp());
ret.setUser(v1Event.getUser());
ret.setDetails(v1Event.getDetails());
ret.setEventKey(v1Event.getEventKey());
ret.setAction(getV2AuditAction(v1Event.getAction()));
AtlasEntitiesWithExtInfo entitiesWithExtInfo = toAtlasEntity(v1Event.getEntityDefinition());
if (entitiesWithExtInfo != null && CollectionUtils.isNotEmpty(entitiesWithExtInfo.getEntities())) {
// there will only one source entity
AtlasEntity entity = entitiesWithExtInfo.getEntities().get(0);
ret.setEntity(entity);
}
return ret;
}
private EntityAuditEvent.EntityAuditAction getV1AuditAction(EntityAuditEventV2.EntityAuditAction v2AuditAction) {
EntityAuditEvent.EntityAuditAction ret = null;
switch (v2AuditAction) {
case ENTITY_CREATE:
case ENTITY_UPDATE:
case ENTITY_DELETE:
case ENTITY_IMPORT_CREATE:
case ENTITY_IMPORT_UPDATE:
case ENTITY_IMPORT_DELETE:
ret = EntityAuditEvent.EntityAuditAction.valueOf(v2AuditAction.name());
break;
case CLASSIFICATION_ADD:
ret = EntityAuditEvent.EntityAuditAction.valueOf(TAG_ADD.name());
break;
case CLASSIFICATION_DELETE:
ret = EntityAuditEvent.EntityAuditAction.valueOf(TAG_DELETE.name());
break;
case CLASSIFICATION_UPDATE:
ret = EntityAuditEvent.EntityAuditAction.valueOf(TAG_UPDATE.name());
break;
}
return ret;
}
private EntityAuditEventV2.EntityAuditAction getV2AuditAction(EntityAuditEvent.EntityAuditAction v1AuditAction) {
EntityAuditEventV2.EntityAuditAction ret = null;
switch (v1AuditAction) {
case ENTITY_CREATE:
case ENTITY_UPDATE:
case ENTITY_DELETE:
case ENTITY_IMPORT_CREATE:
case ENTITY_IMPORT_UPDATE:
case ENTITY_IMPORT_DELETE:
ret = EntityAuditEventV2.EntityAuditAction.valueOf(v1AuditAction.name());
break;
case TAG_ADD:
ret = EntityAuditEventV2.EntityAuditAction.valueOf(CLASSIFICATION_ADD.name());
break;
case TAG_DELETE:
ret = EntityAuditEventV2.EntityAuditAction.valueOf(CLASSIFICATION_DELETE.name());
break;
case TAG_UPDATE:
ret = EntityAuditEventV2.EntityAuditAction.valueOf(CLASSIFICATION_UPDATE.name());
break;
}
return ret;
}
}
\ No newline at end of file
repository/src/main/java/org/apache/atlas/repository/graph/GraphBackedSearchIndexer.java
......@@ -74,6 +74,7 @@ import static org.apache.atlas.repository.Constants.FULLTEXT_INDEX;
import static org.apache.atlas.repository.Constants.GUID_PROPERTY_KEY;
import static org.apache.atlas.repository.Constants.MODIFICATION_TIMESTAMP_PROPERTY_KEY;
import static org.apache.atlas.repository.Constants.MODIFIED_BY_KEY;
import static org.apache.atlas.repository.Constants.PROPAGATED_TRAIT_NAMES_PROPERTY_KEY;
import static org.apache.atlas.repository.Constants.RELATIONSHIP_GUID_PROPERTY_KEY;
import static org.apache.atlas.repository.Constants.STATE_PROPERTY_KEY;
import static org.apache.atlas.repository.Constants.SUPER_TYPES_PROPERTY_KEY;
......@@ -275,6 +276,7 @@ public class GraphBackedSearchIndexer implements SearchIndexer, ActiveStateChang
createVertexIndex(management, ENTITY_TYPE_PROPERTY_KEY, String.class, false, SINGLE, true, true);
createVertexIndex(management, SUPER_TYPES_PROPERTY_KEY, String.class, false, SET, true, true);
createVertexIndex(management, TRAIT_NAMES_PROPERTY_KEY, String.class, false, SET, true, true);
createVertexIndex(management, PROPAGATED_TRAIT_NAMES_PROPERTY_KEY, String.class, false, LIST, true, true);
createVertexIndex(management, TYPENAME_PROPERTY_KEY, String.class, true, SINGLE, true, true);
createVertexIndex(management, VERTEX_TYPE_PROPERTY_KEY, String.class, false, SINGLE, true, true);
......
repository/src/main/java/org/apache/atlas/repository/store/graph/v1/AtlasEntityStoreV1.java
......@@ -462,54 +462,26 @@ public class AtlasEntityStoreV1 implements AtlasEntityStore {
validateEntityAssociations(guid, classifications);
entityGraphMapper.addClassifications(new EntityMutationContext(), guid, classifications);
// notify listeners on classification addition
entityChangeNotifier.onClassificationAddedToEntity(guid, classifications);
}
@Override
@GraphTransaction
public void updateClassifications(String guid, List<AtlasClassification> newClassifications) throws AtlasBaseException {
public void updateClassifications(String guid, List<AtlasClassification> classifications) throws AtlasBaseException {
if (LOG.isDebugEnabled()) {
LOG.debug("Updating classifications={} for entity={}", newClassifications, guid);
LOG.debug("Updating classifications={} for entity={}", classifications, guid);
}
if (StringUtils.isEmpty(guid)) {
throw new AtlasBaseException(AtlasErrorCode.INVALID_PARAMETERS, "Guid not specified");
}
if (CollectionUtils.isEmpty(newClassifications)) {
if (CollectionUtils.isEmpty(classifications)) {
throw new AtlasBaseException(AtlasErrorCode.INVALID_PARAMETERS, "classifications(s) not specified");
}
GraphTransactionInterceptor.lockObjectAndReleasePostCommit(guid);
List<AtlasClassification> updatedClassifications = new ArrayList<>();
for (AtlasClassification newClassification : newClassifications) {
String classificationName = newClassification.getTypeName();
AtlasClassification oldClassification = getClassification(guid, classificationName);
if (oldClassification == null) {
throw new AtlasBaseException(AtlasErrorCode.CLASSIFICATION_NOT_FOUND, classificationName);
}
validateAndNormalizeForUpdate(newClassification);
Map<String, Object> newAttrs = newClassification.getAttributes();
if (MapUtils.isNotEmpty(newAttrs)) {
for (String attrName : newAttrs.keySet()) {
oldClassification.setAttribute(attrName, newAttrs.get(attrName));
}
}
entityGraphMapper.updateClassification(new EntityMutationContext(), guid, oldClassification);
updatedClassifications.add(oldClassification);
}
// notify listeners on update to classifications
entityChangeNotifier.onClassificationUpdatedToEntity(guid, updatedClassifications);
entityGraphMapper.updateClassifications(new EntityMutationContext(), guid, classifications);
}
@Override
......@@ -533,15 +505,10 @@ public class AtlasEntityStoreV1 implements AtlasEntityStore {
List<AtlasClassification> classifications = Collections.singletonList(classification);
for (String guid : guids) {
// validate if entity, not already associated with classifications
validateEntityAssociations(guid, classifications);
entityGraphMapper.addClassifications(new EntityMutationContext(), guid, classifications);
// notify listeners on classification addition
entityChangeNotifier.onClassificationAddedToEntity(guid, classifications);
}
}
@Override
......@@ -561,16 +528,13 @@ public class AtlasEntityStoreV1 implements AtlasEntityStore {
GraphTransactionInterceptor.lockObjectAndReleasePostCommit(guid);
entityGraphMapper.deleteClassifications(guid, classificationNames);
// notify listeners on classification deletion
entityChangeNotifier.onClassificationDeletedFromEntity(guid, classificationNames);
}
@Override
@GraphTransaction
public List<AtlasClassification> getClassifications(String guid) throws AtlasBaseException {
if (LOG.isDebugEnabled()) {
LOG.debug("Getting classifications for entities={}", guid);
LOG.debug("Getting classifications for entity={}", guid);
}
EntityGraphRetriever graphRetriever = new EntityGraphRetriever(typeRegistry);
......@@ -680,24 +644,6 @@ public class AtlasEntityStoreV1 implements AtlasEntityStore {
type.getNormalizedValue(classification);
}
private void validateAndNormalizeForUpdate(AtlasClassification classification) throws AtlasBaseException {
AtlasClassificationType type = typeRegistry.getClassificationTypeByName(classification.getTypeName());
if (type == null) {
throw new AtlasBaseException(AtlasErrorCode.CLASSIFICATION_NOT_FOUND, classification.getTypeName());
}
List<String> messages = new ArrayList<>();
type.validateValueForUpdate(classification, classification.getTypeName(), messages);
if (!messages.isEmpty()) {
throw new AtlasBaseException(AtlasErrorCode.INVALID_PARAMETERS, messages);
}
type.getNormalizedValueForUpdate(classification);
}
/**
* Validate if classification is not already associated with the entities
*
......@@ -734,7 +680,11 @@ public class AtlasEntityStoreV1 implements AtlasEntityStore {
ret = new ArrayList<>();
for (AtlasClassification classification : classifications) {
ret.add(classification.getTypeName());
String entityGuid = classification.getEntityGuid();
if (StringUtils.isEmpty(entityGuid) || StringUtils.equalsIgnoreCase(guid, entityGuid)) {
ret.add(classification.getTypeName());
}
}
}
......
repository/src/main/java/org/apache/atlas/repository/store/graph/v1/AtlasRelationshipStoreV1.java
......@@ -21,6 +21,7 @@ import org.apache.atlas.AtlasErrorCode;
import org.apache.atlas.annotation.GraphTransaction;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.TypeCategory;
import org.apache.atlas.model.instance.AtlasEntity.Status;
import org.apache.atlas.model.instance.AtlasObjectId;
import org.apache.atlas.model.instance.AtlasRelationship;
import org.apache.atlas.model.typedef.AtlasRelationshipDef;
......@@ -30,6 +31,7 @@ import org.apache.atlas.repository.Constants;
import org.apache.atlas.repository.RepositoryException;
import org.apache.atlas.repository.graph.GraphHelper;
import org.apache.atlas.repository.graphdb.AtlasEdge;
import org.apache.atlas.repository.graphdb.AtlasEdgeDirection;
import org.apache.atlas.repository.graphdb.AtlasVertex;
import org.apache.atlas.repository.store.graph.AtlasRelationshipStore;
import org.apache.atlas.type.AtlasEntityType;
......@@ -54,10 +56,16 @@ import java.util.Objects;
import java.util.Set;
import java.util.UUID;
import static org.apache.atlas.model.instance.AtlasEntity.Status.ACTIVE;
import static org.apache.atlas.model.instance.AtlasEntity.Status.DELETED;
import static org.apache.atlas.model.typedef.AtlasRelationshipDef.PropagateTags.NONE;
import static org.apache.atlas.model.typedef.AtlasRelationshipDef.PropagateTags.BOTH;
import static org.apache.atlas.model.typedef.AtlasRelationshipDef.PropagateTags.ONE_TO_TWO;
import static org.apache.atlas.model.typedef.AtlasRelationshipDef.PropagateTags.TWO_TO_ONE;
import static org.apache.atlas.repository.graphdb.AtlasEdgeDirection.BOTH;
import static org.apache.atlas.repository.graph.GraphHelper.getGuid;
import static org.apache.atlas.repository.graph.GraphHelper.getOutGoingEdgesByLabel;
import static org.apache.atlas.repository.graph.GraphHelper.getPropagateTags;
import static org.apache.atlas.repository.store.graph.v1.AtlasGraphUtilsV1.getIdFromVertex;
import static org.apache.atlas.repository.store.graph.v1.AtlasGraphUtilsV1.getState;
import static org.apache.atlas.repository.store.graph.v1.AtlasGraphUtilsV1.getTypeName;
......@@ -117,8 +125,8 @@ public class AtlasRelationshipStoreV1 implements AtlasRelationshipStore {
AtlasEdge edge = graphHelper.getEdgeForGUID(guid);
String edgeType = AtlasGraphUtilsV1.getTypeName(edge);
AtlasVertex end1Vertex = edge.getInVertex();
AtlasVertex end2Vertex = edge.getOutVertex();
AtlasVertex end1Vertex = edge.getOutVertex();
AtlasVertex end2Vertex = edge.getInVertex();
// update shouldn't change endType
if (StringUtils.isNotEmpty(relationship.getTypeName()) && !StringUtils.equalsIgnoreCase(edgeType, relationship.getTypeName())) {
......@@ -302,6 +310,8 @@ public class AtlasRelationshipStoreV1 implements AtlasRelationshipStore {
private AtlasRelationship updateRelationship(AtlasEdge relationshipEdge, AtlasRelationship relationship) throws AtlasBaseException {
AtlasRelationshipType relationType = typeRegistry.getRelationshipTypeByName(relationship.getTypeName());
updateTagPropagations(relationshipEdge, relationship.getPropagateTags());
AtlasGraphUtilsV1.setProperty(relationshipEdge, Constants.RELATIONSHIPTYPE_TAG_PROPAGATION_KEY, relationship.getPropagateTags().name());
if (MapUtils.isNotEmpty(relationType.getAllAttributes())) {
......@@ -318,6 +328,46 @@ public class AtlasRelationshipStoreV1 implements AtlasRelationshipStore {
return entityRetriever.mapEdgeToAtlasRelationship(relationshipEdge);
}
private void updateTagPropagations(AtlasEdge relationshipEdge, PropagateTags tagPropagation) throws AtlasBaseException {
PropagateTags oldTagPropagation = getPropagateTags(relationshipEdge);
PropagateTags newTagPropagation = tagPropagation;
if (newTagPropagation != oldTagPropagation) {
if (LOG.isDebugEnabled()) {
LOG.debug("Updating tagPropagation property: [ {} -> {} ] for relationship: [{} --> {}]", oldTagPropagation.name(),
newTagPropagation.name(), getTypeName(relationshipEdge.getOutVertex()), getTypeName(relationshipEdge.getInVertex()));
}
if (oldTagPropagation == NONE) {
entityRetriever.addTagPropagation(relationshipEdge, newTagPropagation);
} else if (oldTagPropagation == ONE_TO_TWO) {
if (newTagPropagation == NONE || newTagPropagation == TWO_TO_ONE) {
entityRetriever.removeTagPropagation(relationshipEdge, oldTagPropagation);
}
if (newTagPropagation != NONE) {
entityRetriever.addTagPropagation(relationshipEdge, newTagPropagation);
}
} else if (oldTagPropagation == TWO_TO_ONE) {
if (newTagPropagation == NONE || newTagPropagation == ONE_TO_TWO) {
entityRetriever.removeTagPropagation(relationshipEdge, oldTagPropagation);
}
if (newTagPropagation != NONE) {
entityRetriever.addTagPropagation(relationshipEdge, newTagPropagation);
}
} else if (oldTagPropagation == BOTH) {
if (newTagPropagation == ONE_TO_TWO || newTagPropagation == NONE) {
entityRetriever.removeTagPropagation(relationshipEdge, TWO_TO_ONE);
}
if (newTagPropagation == TWO_TO_ONE || newTagPropagation == NONE) {
entityRetriever.removeTagPropagation(relationshipEdge, ONE_TO_TWO);
}
}
}
}
private void validateRelationship(AtlasRelationship relationship) throws AtlasBaseException {
if (relationship == null) {
throw new AtlasBaseException(AtlasErrorCode.INVALID_PARAMETERS, "AtlasRelationship is null");
......@@ -462,16 +512,20 @@ public class AtlasRelationshipStoreV1 implements AtlasRelationshipStore {
}
public AtlasEdge getRelationshipEdge(AtlasVertex fromVertex, AtlasVertex toVertex, String relationshipType) {
String relationshipLabel = getRelationshipEdgeLabel(fromVertex, toVertex, relationshipType);
AtlasEdge ret = graphHelper.getEdgeForLabel(fromVertex, relationshipLabel);
String relationshipLabel = getRelationshipEdgeLabel(fromVertex, toVertex, relationshipType);
Iterator<AtlasEdge> edgesIterator = getOutGoingEdgesByLabel(fromVertex, relationshipLabel);
AtlasEdge ret = null;
if (ret != null) {
AtlasVertex inVertex = ret.getInVertex();
while (edgesIterator != null && edgesIterator.hasNext()) {
AtlasEdge edge = edgesIterator.next();
if (inVertex != null) {
if (!StringUtils.equals(AtlasGraphUtilsV1.getIdFromVertex(inVertex),
AtlasGraphUtilsV1.getIdFromVertex(toVertex))) {
ret = null;
if (edge != null) {
Status status = graphHelper.getStatus(edge);
if ((status == null || status == ACTIVE) &&
StringUtils.equals(getIdFromVertex(edge.getInVertex()), getIdFromVertex(toVertex))) {
ret = edge;
break;
}
}
}
......@@ -499,11 +553,15 @@ public class AtlasRelationshipStoreV1 implements AtlasRelationshipStore {
return ret;
}
private AtlasEdge createRelationshipEdge(AtlasVertex fromVertex, AtlasVertex toVertex, AtlasRelationship relationship) throws RepositoryException {
private AtlasEdge createRelationshipEdge(AtlasVertex fromVertex, AtlasVertex toVertex, AtlasRelationship relationship) throws RepositoryException, AtlasBaseException {
String relationshipLabel = getRelationshipEdgeLabel(fromVertex, toVertex, relationship.getTypeName());
PropagateTags tagPropagation = getRelationshipTagPropagation(fromVertex, toVertex, relationship);
AtlasEdge ret = graphHelper.getOrCreateEdge(fromVertex, toVertex, relationshipLabel);
if (LOG.isDebugEnabled()) {
LOG.debug("Created relationship edge from [{}] --> [{}] using edge label: [{}]", getTypeName(fromVertex), getTypeName(toVertex), relationshipLabel);
}
// map additional properties to relationship edge
if (ret != null) {
final String guid = UUID.randomUUID().toString();
......@@ -512,6 +570,9 @@ public class AtlasRelationshipStoreV1 implements AtlasRelationshipStore {
AtlasGraphUtilsV1.setProperty(ret, Constants.RELATIONSHIP_GUID_PROPERTY_KEY, guid);
AtlasGraphUtilsV1.setProperty(ret, Constants.VERSION_PROPERTY_KEY, getRelationshipVersion(relationship));
AtlasGraphUtilsV1.setProperty(ret, Constants.RELATIONSHIPTYPE_TAG_PROPAGATION_KEY, tagPropagation.name());
// propagate tags
entityRetriever.addTagPropagation(ret, tagPropagation);
}
return ret;
......@@ -596,7 +657,7 @@ public class AtlasRelationshipStoreV1 implements AtlasRelationshipStore {
*/
private boolean vertexHasRelationshipWithType(AtlasVertex vertex, String relationshipTypeName) {
String relationshipEdgeLabel = getRelationshipEdgeLabel(getTypeName(vertex), relationshipTypeName);
Iterator<AtlasEdge> iter = graphHelper.getAdjacentEdgesByLabel(vertex, BOTH, relationshipEdgeLabel);
Iterator<AtlasEdge> iter = graphHelper.getAdjacentEdgesByLabel(vertex, AtlasEdgeDirection.BOTH, relationshipEdgeLabel);
return (iter != null) ? iter.hasNext() : false;
}
......
repository/src/main/java/org/apache/atlas/util/AtlasGremlin3QueryProvider.java
......@@ -63,6 +63,18 @@ public class AtlasGremlin3QueryProvider extends AtlasGremlin2QueryProvider {
return "g.V().range(0,1).toList()";
case GREMLIN_SEARCH_RETURNS_EDGE_ID:
return "g.E().range(0,1).toList()";
case TAG_PROPAGATION_IMPACTED_INSTANCES:
return "g.V().has('__guid', guid).aggregate('src')" +
".repeat(union(outE().has('__state', 'ACTIVE').has('tagPropagation', within('ONE_TO_TWO', 'BOTH')).inV(), " +
"inE().has('__state', 'ACTIVE').has('tagPropagation', within('TWO_TO_ONE', 'BOTH')).outV())" +
".dedup().where(without('src')).simplePath()).emit().toList();";
case TAG_PROPAGATION_IMPACTED_INSTANCES_FOR_REMOVAL:
return "g.V().has('__guid', guid).aggregate('src')" +
".repeat(union(outE().has('__state', 'ACTIVE').has('tagPropagation', within('ONE_TO_TWO', 'BOTH')).has('_r__guid', neq(relationshipGuid)).inV(), " +
"inE().has('__state', 'ACTIVE').has('tagPropagation', within('TWO_TO_ONE', 'BOTH')).has('_r__guid', neq(relationshipGuid)).outV())" +
".dedup().where(without('src')).simplePath()).emit().toList();";
}
return super.getQuery(gremlinQuery);
}
......
repository/src/main/java/org/apache/atlas/util/AtlasGremlinQueryProvider.java
......@@ -81,6 +81,9 @@ public abstract class AtlasGremlinQueryProvider {
COMPARE_ENDS_WITH,
COMPARE_CONTAINS,
COMPARE_IS_NULL,
COMPARE_NOT_NULL
COMPARE_NOT_NULL,
TAG_PROPAGATION_IMPACTED_INSTANCES,
TAG_PROPAGATION_IMPACTED_INSTANCES_FOR_REMOVAL
}
}
repository/src/main/java/org/apache/atlas/util/AtlasRepositoryConfiguration.java
......@@ -25,6 +25,7 @@ import org.apache.atlas.repository.graphdb.GraphDatabase;
import org.apache.atlas.repository.store.graph.v1.DeleteHandlerV1;
import org.apache.atlas.repository.store.graph.v1.SoftDeleteHandlerV1;
import org.apache.commons.configuration.Configuration;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
......@@ -52,6 +53,7 @@ public class AtlasRepositoryConfiguration {
private static Integer typeUpdateLockMaxWaitTimeInSeconds = null;
private static final String ENABLE_FULLTEXT_SEARCH_PROPERTY = "atlas.search.fulltext.enable";
private static final String ENTITY_NOTIFICATION_VERSION_PROPERTY = "atlas.notification.entity.version";
/**
* Configures whether the full text vertex property is populated. Turning this off
......@@ -62,6 +64,19 @@ public class AtlasRepositoryConfiguration {
return ApplicationProperties.get().getBoolean(ENABLE_FULLTEXT_SEARCH_PROPERTY, true);
}
public static boolean isV2EntityNotificationEnabled() {
boolean ret;
try {
String notificationVersion = ApplicationProperties.get().getString(ENTITY_NOTIFICATION_VERSION_PROPERTY, "v2");
return StringUtils.equalsIgnoreCase(notificationVersion, "v2");
} catch (AtlasException e) {
ret = true;
}
return ret;
}
private static final String AUDIT_REPOSITORY_IMPLEMENTATION_PROPERTY = "atlas.EntityAuditRepository.impl";
@SuppressWarnings("unchecked")
......
repository/src/test/java/org/apache/atlas/TestModules.java
......@@ -30,8 +30,10 @@ import org.apache.atlas.discovery.EntityDiscoveryService;
import org.apache.atlas.discovery.EntityLineageService;
import org.apache.atlas.graph.GraphSandboxUtil;
import org.apache.atlas.listener.EntityChangeListener;
import org.apache.atlas.listener.EntityChangeListenerV2;
import org.apache.atlas.listener.TypeDefChangeListener;
import org.apache.atlas.repository.audit.EntityAuditListener;
import org.apache.atlas.repository.audit.EntityAuditListenerV2;
import org.apache.atlas.repository.audit.EntityAuditRepository;
import org.apache.atlas.repository.graph.GraphBackedSearchIndexer;
import org.apache.atlas.repository.graphdb.AtlasGraph;
......@@ -145,6 +147,10 @@ public class TestModules {
Multibinder.newSetBinder(binder(), EntityChangeListener.class);
entityChangeListenerBinder.addBinding().to(EntityAuditListener.class);
Multibinder<EntityChangeListenerV2> entityChangeListenerV2Binder =
Multibinder.newSetBinder(binder(), EntityChangeListenerV2.class);
entityChangeListenerV2Binder.addBinding().to(EntityAuditListenerV2.class);
final GraphTransactionInterceptor graphTransactionInterceptor = new GraphTransactionInterceptor(new AtlasGraphProvider().get());
requestInjection(graphTransactionInterceptor);
bindInterceptor(Matchers.any(), Matchers.annotatedWith(GraphTransaction.class), graphTransactionInterceptor);
......
repository/src/test/java/org/apache/atlas/repository/audit/AuditRepositoryTestBase.java
......@@ -46,9 +46,9 @@ public class AuditRepositoryTestBase {
EntityAuditEvent event = new EntityAuditEvent(rand(), System.currentTimeMillis(), "u1",
EntityAuditEvent.EntityAuditAction.ENTITY_CREATE, "d1", new Referenceable(rand()));
eventRepository.putEvents(event);
eventRepository.putEventsV1(event);
List<EntityAuditEvent> events = eventRepository.listEvents(event.getEntityId(), null, (short) 10);
List<EntityAuditEvent> events = eventRepository.listEventsV1(event.getEntityId(), null, (short) 10);
assertEquals(events.size(), 1);
assertEventEquals(events.get(0), event);
......@@ -67,28 +67,28 @@ public class AuditRepositoryTestBase {
//Add events for both ids
EntityAuditEvent event = new EntityAuditEvent(id2, ts - i, "user" + i, EntityAuditEvent.EntityAuditAction.ENTITY_UPDATE, "details" + i, entity);
eventRepository.putEvents(event);
eventRepository.putEventsV1(event);
expectedEvents.add(event);
eventRepository.putEvents(new EntityAuditEvent(id1, ts - i, "user" + i, EntityAuditEvent.EntityAuditAction.TAG_DELETE, "details" + i, entity));
eventRepository.putEvents(new EntityAuditEvent(id3, ts - i, "user" + i, EntityAuditEvent.EntityAuditAction.TAG_ADD, "details" + i, entity));
eventRepository.putEventsV1(new EntityAuditEvent(id1, ts - i, "user" + i, EntityAuditEvent.EntityAuditAction.TAG_DELETE, "details" + i, entity));
eventRepository.putEventsV1(new EntityAuditEvent(id3, ts - i, "user" + i, EntityAuditEvent.EntityAuditAction.TAG_ADD, "details" + i, entity));
}
//Use ts for which there is no event - ts + 2
List<EntityAuditEvent> events = eventRepository.listEvents(id2, null, (short) 3);
List<EntityAuditEvent> events = eventRepository.listEventsV1(id2, null, (short) 3);
assertEquals(events.size(), 3);
assertEventEquals(events.get(0), expectedEvents.get(0));
assertEventEquals(events.get(1), expectedEvents.get(1));
assertEventEquals(events.get(2), expectedEvents.get(2));
//Use last event's timestamp for next list(). Should give only 1 event and shouldn't include events from other id
events = eventRepository.listEvents(id2, events.get(2).getEventKey(), (short) 3);
events = eventRepository.listEventsV1(id2, events.get(2).getEventKey(), (short) 3);
assertEquals(events.size(), 1);
assertEventEquals(events.get(0), expectedEvents.get(2));
}
@Test
public void testInvalidEntityId() throws Exception {
List<EntityAuditEvent> events = eventRepository.listEvents(rand(), null, (short) 3);
List<EntityAuditEvent> events = eventRepository.listEventsV1(rand(), null, (short) 3);
assertEquals(events.size(), 0);
}
......
webapp/src/main/java/org/apache/atlas/notification/EntityNotificationListenerV2.java 0 → 100644
/**
* 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.atlas.notification;
import org.apache.atlas.ApplicationProperties;
import org.apache.atlas.AtlasErrorCode;
import org.apache.atlas.AtlasException;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.listener.EntityChangeListenerV2;
import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntity;
import org.apache.atlas.type.AtlasClassificationType;
import org.apache.atlas.type.AtlasTypeRegistry;
import org.apache.atlas.v1.model.notification.EntityNotificationV2;
import org.apache.atlas.v1.model.notification.EntityNotificationV2.OperationType;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.configuration.Configuration;
import org.springframework.stereotype.Component;
import javax.inject.Inject;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import static org.apache.atlas.notification.NotificationEntityChangeListener.ATLAS_ENTITY_NOTIFICATION_PROPERTY;
import static org.apache.atlas.notification.NotificationInterface.NotificationType.ENTITIES;
import static org.apache.atlas.v1.model.notification.EntityNotificationV2.OperationType.CLASSIFICATION_ADD;
import static org.apache.atlas.v1.model.notification.EntityNotificationV2.OperationType.CLASSIFICATION_DELETE;
import static org.apache.atlas.v1.model.notification.EntityNotificationV2.OperationType.CLASSIFICATION_UPDATE;
import static org.apache.atlas.repository.graph.GraphHelper.isInternalType;
import static org.apache.atlas.v1.model.notification.EntityNotificationV2.OperationType.ENTITY_CREATE;
import static org.apache.atlas.v1.model.notification.EntityNotificationV2.OperationType.ENTITY_DELETE;
import static org.apache.atlas.v1.model.notification.EntityNotificationV2.OperationType.ENTITY_UPDATE;
@Component
public class EntityNotificationListenerV2 implements EntityChangeListenerV2 {
private final AtlasTypeRegistry typeRegistry;
private final NotificationInterface notificationInterface;
private final Map<String, List<String>> notificationAttributesCache = new HashMap<>();
private static Configuration APPLICATION_PROPERTIES = null;
@Inject
public EntityNotificationListenerV2(AtlasTypeRegistry typeRegistry, NotificationInterface notificationInterface) {
this.typeRegistry = typeRegistry;
this.notificationInterface = notificationInterface;
}
@Override
public void onEntitiesAdded(List<AtlasEntity> entities, boolean isImport) throws AtlasBaseException {
notifyEntityEvents(entities, ENTITY_CREATE);
}
@Override
public void onEntitiesUpdated(List<AtlasEntity> entities, boolean isImport) throws AtlasBaseException {
notifyEntityEvents(entities, ENTITY_UPDATE);
}
@Override
public void onEntitiesDeleted(List<AtlasEntity> entities, boolean isImport) throws AtlasBaseException {
notifyEntityEvents(entities, ENTITY_DELETE);
}
@Override
public void onClassificationsAdded(AtlasEntity entity, List<AtlasClassification> classifications) throws AtlasBaseException {
notifyEntityEvents(Collections.singletonList(entity), CLASSIFICATION_ADD);
}
@Override
public void onClassificationsUpdated(AtlasEntity entity, List<AtlasClassification> classifications) throws AtlasBaseException {
notifyEntityEvents(Collections.singletonList(entity), CLASSIFICATION_UPDATE);
}
@Override
public void onClassificationsDeleted(AtlasEntity entity, List<String> classificationNames) throws AtlasBaseException {
notifyEntityEvents(Collections.singletonList(entity), CLASSIFICATION_DELETE);
}
private void notifyEntityEvents(List<AtlasEntity> entities, OperationType operationType) throws AtlasBaseException {
List<EntityNotificationV2> messages = new ArrayList<>();
for (AtlasEntity entity : entities) {
if (isInternalType(entity.getTypeName())) {
continue;
}
filterNotificationAttributes(entity);
messages.add(new EntityNotificationV2(entity, operationType, getAllClassifications(entity)));
}
if (!messages.isEmpty()) {
try {
notificationInterface.send(ENTITIES, messages);
} catch (NotificationException e) {
throw new AtlasBaseException(AtlasErrorCode.ENTITY_NOTIFICATION_FAILED, e, operationType.name());
}
}
}
private List<AtlasClassification> getAllClassifications(AtlasEntity entity) {
List<AtlasClassification> ret = getAllClassifications(entity.getClassifications(), typeRegistry);
return ret;
}
private static List<AtlasClassification> getAllClassifications(List<AtlasClassification> classifications, AtlasTypeRegistry typeRegistry) {
List<AtlasClassification> ret = new LinkedList<>();
if (CollectionUtils.isNotEmpty(classifications)) {
for (AtlasClassification classification : classifications) {
AtlasClassificationType classificationType = typeRegistry.getClassificationTypeByName(classification.getTypeName());
Set<String> superTypeNames = classificationType != null ? classificationType.getAllSuperTypes() : null;
ret.add(classification);
if (CollectionUtils.isNotEmpty(superTypeNames)) {
for (String superTypeName : superTypeNames) {
AtlasClassification superTypeClassification = new AtlasClassification(superTypeName);
superTypeClassification.setEntityGuid(classification.getEntityGuid());
superTypeClassification.setPropagate(classification.isPropagate());
if (MapUtils.isNotEmpty(classification.getAttributes())) {
AtlasClassificationType superType = typeRegistry.getClassificationTypeByName(superTypeName);
if (superType != null && MapUtils.isNotEmpty(superType.getAllAttributes())) {
Map<String, Object> superTypeClassificationAttributes = new HashMap<>();
for (Map.Entry<String, Object> attrEntry : classification.getAttributes().entrySet()) {
String attrName = attrEntry.getKey();
if (superType.getAllAttributes().containsKey(attrName)) {
superTypeClassificationAttributes.put(attrName, attrEntry.getValue());
}
}
superTypeClassification.setAttributes(superTypeClassificationAttributes);
}
}
ret.add(superTypeClassification);
}
}
}
}
return ret;
}
private void filterNotificationAttributes(AtlasEntity entity) {
Map<String, Object> attributesMap = entity.getAttributes();
List<String> notificationAttrs = getNotificationAttributes(entity.getTypeName());
if (MapUtils.isNotEmpty(attributesMap) && CollectionUtils.isNotEmpty(notificationAttrs)) {
Collection<String> attributesToRemove = CollectionUtils.subtract(attributesMap.keySet(), notificationAttrs);
for (String attributeToRemove : attributesToRemove) {
attributesMap.remove(attributeToRemove);
}
}
}
private List<String> getNotificationAttributes(String entityType) {
List<String> ret = null;
initApplicationProperties();
if (notificationAttributesCache.containsKey(entityType)) {
ret = notificationAttributesCache.get(entityType);
} else if (APPLICATION_PROPERTIES != null) {
String[] notificationAttributes = APPLICATION_PROPERTIES.getStringArray(ATLAS_ENTITY_NOTIFICATION_PROPERTY + "." +
entityType + "." + "attributes.include");
if (notificationAttributes != null) {
ret = Arrays.asList(notificationAttributes);
}
notificationAttributesCache.put(entityType, ret);
}
return ret;
}
private void initApplicationProperties() {
if (APPLICATION_PROPERTIES == null) {
try {
APPLICATION_PROPERTIES = ApplicationProperties.get();
} catch (AtlasException ex) {
// ignore
}
}
}
}
\ No newline at end of file
webapp/src/main/java/org/apache/atlas/notification/NotificationEntityChangeListener.java
......@@ -42,7 +42,7 @@ import java.util.*;
*/
@Component
public class NotificationEntityChangeListener implements EntityChangeListener {
private static final String ATLAS_ENTITY_NOTIFICATION_PROPERTY = "atlas.notification.entity";
protected static final String ATLAS_ENTITY_NOTIFICATION_PROPERTY = "atlas.notification.entity";
private final NotificationInterface notificationInterface;
private final AtlasTypeRegistry typeRegistry;
......
webapp/src/main/java/org/apache/atlas/web/resources/EntityResource.java
......@@ -18,7 +18,6 @@
package org.apache.atlas.web.resources;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
......@@ -28,7 +27,7 @@ import org.apache.atlas.AtlasErrorCode;
import org.apache.atlas.AtlasException;
import org.apache.atlas.CreateUpdateEntitiesResult;
import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.discovery.AtlasDiscoveryService;
import org.apache.atlas.model.audit.EntityAuditEventV2;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.instance.AtlasClassification;
import org.apache.atlas.model.instance.AtlasEntity.AtlasEntitiesWithExtInfo;
......@@ -98,7 +97,7 @@ public class EntityResource {
private final AtlasTypeRegistry typeRegistry;
private final EntityREST entityREST;
private final EntityAuditRepository entityAuditRepository;
private final AtlasDiscoveryService atlasDiscoveryService;
private final AtlasInstanceConverter instanceConverter;
@Context
UriInfo uriInfo;
......@@ -109,13 +108,13 @@ public class EntityResource {
final AtlasTypeRegistry typeRegistry,
final EntityREST entityREST,
final EntityAuditRepository entityAuditRepository,
final AtlasDiscoveryService atlasDiscoveryService) {
final AtlasInstanceConverter instanceConverter) {
this.restAdapters = restAdapters;
this.entitiesStore = entitiesStore;
this.typeRegistry = typeRegistry;
this.entityREST = entityREST;
this.entityAuditRepository = entityAuditRepository;
this.atlasDiscoveryService = atlasDiscoveryService;
this.instanceConverter = instanceConverter;
}
/**
......@@ -1133,20 +1132,27 @@ public class EntityResource {
AtlasPerfTracer perf = null;
if (LOG.isDebugEnabled()) {
LOG.debug("Audit events request for entity {}, start key {}, number of results required {}", guid, startKey, count);
}
try {
if (AtlasPerfTracer.isPerfTraceEnabled(PERF_LOG)) {
perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "EntityResource.getAuditEvents(" + guid + ", " + startKey + ", " + count + ")");
}
List<EntityAuditEvent> events = entityAuditRepository.listEvents(guid, startKey, count);
List events = entityAuditRepository.listEvents(guid, startKey, count);
List<EntityAuditEvent> v1Events = new ArrayList<>();
for (Object event : events) {
if (event instanceof EntityAuditEvent) {
v1Events.add((EntityAuditEvent) event);
} else if (event instanceof EntityAuditEventV2) {
v1Events.add(instanceConverter.toV1AuditEvent((EntityAuditEventV2) event));
} else {
LOG.warn("unknown entity-audit event type {}. Ignored", event != null ? event.getClass().getCanonicalName() : "null");
}
}
Map<String, Object> response = new HashMap<>();
response.put(AtlasClient.REQUEST_ID, Servlets.getRequestId());
response.put(AtlasClient.EVENTS, events);
response.put(AtlasClient.EVENTS, v1Events);
return Response.ok(AtlasJson.toV1Json(response)).build();
} catch (IllegalArgumentException e) {
LOG.error("Unable to get audit events for entity guid={} startKey={}", guid, startKey, e);
......
webapp/src/main/java/org/apache/atlas/web/rest/EntityREST.java
......@@ -18,6 +18,8 @@
package org.apache.atlas.web.rest;
import org.apache.atlas.AtlasErrorCode;
import org.apache.atlas.EntityAuditEvent;
import org.apache.atlas.model.audit.EntityAuditEventV2;
import org.apache.atlas.exception.AtlasBaseException;
import org.apache.atlas.model.TypeCategory;
import org.apache.atlas.model.instance.AtlasClassification;
......@@ -26,6 +28,8 @@ import org.apache.atlas.model.instance.AtlasEntity.AtlasEntityWithExtInfo;
import org.apache.atlas.model.instance.ClassificationAssociateRequest;
import org.apache.atlas.model.instance.EntityMutationResponse;
import org.apache.atlas.model.typedef.AtlasStructDef.AtlasAttributeDef;
import org.apache.atlas.repository.audit.EntityAuditRepository;
import org.apache.atlas.repository.converters.AtlasInstanceConverter;
import org.apache.atlas.repository.store.graph.AtlasEntityStore;
import org.apache.atlas.repository.store.graph.v1.AtlasEntityStream;
import org.apache.atlas.repository.store.graph.v1.EntityStream;
......@@ -38,6 +42,7 @@ import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import javax.inject.Inject;
......@@ -45,6 +50,7 @@ import javax.inject.Singleton;
import javax.servlet.http.HttpServletRequest;
import javax.ws.rs.Consumes;
import javax.ws.rs.DELETE;
import javax.ws.rs.DefaultValue;
import javax.ws.rs.GET;
import javax.ws.rs.POST;
import javax.ws.rs.PUT;
......@@ -67,17 +73,24 @@ import java.util.Map;
@Singleton
@Service
public class EntityREST {
private static final Logger LOG = LoggerFactory.getLogger(EntityREST.class);
private static final Logger PERF_LOG = AtlasPerfTracer.getPerfLogger("rest.EntityREST");
public static final String PREFIX_ATTR = "attr:";
private final AtlasTypeRegistry typeRegistry;
private final AtlasEntityStore entitiesStore;
private final AtlasTypeRegistry typeRegistry;
private final AtlasEntityStore entitiesStore;
private final EntityAuditRepository auditRepository;
private final AtlasInstanceConverter instanceConverter;
@Inject
public EntityREST(AtlasTypeRegistry typeRegistry, AtlasEntityStore entitiesStore) {
this.typeRegistry = typeRegistry;
this.entitiesStore = entitiesStore;
public EntityREST(AtlasTypeRegistry typeRegistry, AtlasEntityStore entitiesStore,
EntityAuditRepository auditRepository, AtlasInstanceConverter instanceConverter) {
this.typeRegistry = typeRegistry;
this.entitiesStore = entitiesStore;
this.auditRepository = auditRepository;
this.instanceConverter = instanceConverter;
}
/**
......@@ -409,14 +422,14 @@ public class EntityREST {
@PUT
@Path("/guid/{guid}/classifications")
@Produces(Servlets.JSON_MEDIA_TYPE)
public void updateClassification(@PathParam("guid") final String guid, List<AtlasClassification> classifications) throws AtlasBaseException {
public void updateClassifications(@PathParam("guid") final String guid, List<AtlasClassification> classifications) throws AtlasBaseException {
Servlets.validateQueryParamLength("guid", guid);
AtlasPerfTracer perf = null;
try {
if (AtlasPerfTracer.isPerfTraceEnabled(PERF_LOG)) {
perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "EntityREST.updateClassification(" + guid + ")");
perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "EntityREST.updateClassifications(" + guid + ")");
}
if (StringUtils.isEmpty(guid)) {
......@@ -581,6 +594,38 @@ public class EntityREST {
}
}
@GET
@Path("{guid}/audit")
@Consumes(Servlets.JSON_MEDIA_TYPE)
@Produces(Servlets.JSON_MEDIA_TYPE)
public List<EntityAuditEventV2> getAuditEvents(@PathParam("guid") String guid, @QueryParam("startKey") String startKey,
@QueryParam("count") @DefaultValue("100") short count) throws AtlasBaseException {
AtlasPerfTracer perf = null;
try {
if (AtlasPerfTracer.isPerfTraceEnabled(PERF_LOG)) {
perf = AtlasPerfTracer.getPerfTracer(PERF_LOG, "EntityREST.getAuditEvents(" + guid + ", " + startKey + ", " + count + ")");
}
List events = auditRepository.listEvents(guid, startKey, count);
List<EntityAuditEventV2> ret = new ArrayList<>();
for (Object event : events) {
if (event instanceof EntityAuditEventV2) {
ret.add((EntityAuditEventV2) event);
} else if (event instanceof EntityAuditEvent) {
ret.add(instanceConverter.toV2AuditEvent((EntityAuditEvent) event));
} else {
LOG.warn("unknown entity-audit event type {}. Ignored", event != null ? event.getClass().getCanonicalName() : "null");
}
}
return ret;
} finally {
AtlasPerfTracer.log(perf);
}
}
private AtlasEntityType ensureEntityType(String typeName) throws AtlasBaseException {
AtlasEntityType ret = typeRegistry.getEntityTypeByName(typeName);
......
webapp/src/test/java/org/apache/atlas/web/adapters/TestEntityREST.java
......@@ -164,7 +164,7 @@ public class TestEntityREST {
put("tag", "tagName_updated");
}});
entityREST.updateClassification(dbEntity.getGuid(), new ArrayList<>(Arrays.asList(phiClassification, testClassification)));
entityREST.updateClassifications(dbEntity.getGuid(), new ArrayList<>(Arrays.asList(phiClassification, testClassification)));
AtlasClassification updatedClassification = entityREST.getClassification(dbEntity.getGuid(), TestUtilsV2.PHI);
Assert.assertNotNull(updatedClassification);
......
webapp/src/test/java/org/apache/atlas/web/integration/EntityJerseyResourceIT.java
......@@ -808,7 +808,7 @@ public class EntityJerseyResourceIT extends BaseResourceIT {
} catch (AtlasServiceException e) {
assertNotNull(e);
assertNotNull(e.getStatus());
assertEquals(e.getStatus(), ClientResponse.Status.NOT_FOUND);
assertEquals(e.getStatus(), ClientResponse.Status.BAD_REQUEST);
}
}
......
webapp/src/test/java/org/apache/atlas/web/integration/EntityV2JerseyResourceIT.java
......@@ -560,7 +560,7 @@ public class EntityV2JerseyResourceIT extends BaseResourceIT {
fail("Deletion should've failed for non-existent trait association");
} catch (AtlasServiceException ex) {
Assert.assertNotNull(ex.getStatus());
assertEquals(ex.getStatus(), ClientResponse.Status.NOT_FOUND);
assertEquals(ex.getStatus(), ClientResponse.Status.BAD_REQUEST);
}
}
......
webapp/src/test/resources/atlas-application.properties
......@@ -64,6 +64,7 @@ atlas.lineage.schema.query.hive_table_v1=hive_table_v1 where __guid='%s'\, colum
######### Notification Configs #########
atlas.notification.embedded=true
atlas.notification.entity.version=v1
atlas.kafka.zookeeper.connect=localhost:19026
atlas.kafka.bootstrap.servers=localhost:19027
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment