repository/src/main/java/org/apache/atlas/repository/audit/EntityAuditListener.java - incubator-atlas - Git at Google

 /**
  * 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.repository.audit;

 import org.apache.atlas.AtlasException;
 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.typesystem.IReferenceableInstance;
 import org.apache.atlas.typesystem.IStruct;
 import org.apache.atlas.typesystem.ITypedReferenceableInstance;
 import org.apache.atlas.typesystem.json.InstanceSerialization;
 import org.apache.atlas.typesystem.types.AttributeInfo;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.collections.MapUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import javax.inject.Inject;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;

 /**
  * Listener on entity create/update/delete, tag add/delete. Adds the corresponding audit event to the audit repository.
  */
 public class EntityAuditListener implements EntityChangeListener {
     private static final Logger LOG = LoggerFactory.getLogger(EntityAuditListener.class);

     private EntityAuditRepository auditRepository;

     @Inject
     public EntityAuditListener(EntityAuditRepository auditRepository) {
         this.auditRepository = auditRepository;
     }

     @Override
     public void onEntitiesAdded(Collection<ITypedReferenceableInstance> entities, boolean isImport) throws AtlasException {
         List<EntityAuditEvent> events = new ArrayList<>();
         for (ITypedReferenceableInstance entity : entities) {
             EntityAuditEvent event = createEvent(entity, isImport ? EntityAuditAction.ENTITY_IMPORT_CREATE : EntityAuditAction.ENTITY_CREATE);
             events.add(event);
         }

         auditRepository.putEvents(events);
     }

     @Override
     public void onEntitiesUpdated(Collection<ITypedReferenceableInstance> entities, boolean isImport) throws AtlasException {
         List<EntityAuditEvent> events = new ArrayList<>();
         for (ITypedReferenceableInstance entity : entities) {
             EntityAuditEvent event = createEvent(entity, isImport ? EntityAuditAction.ENTITY_IMPORT_UPDATE : EntityAuditAction.ENTITY_UPDATE);
             events.add(event);
         }

         auditRepository.putEvents(events);
     }

     @Override
     public void onTraitsAdded(ITypedReferenceableInstance entity, Collection<? extends IStruct> traits) throws AtlasException {
         if (traits != null) {
             for (IStruct trait : traits) {
                 EntityAuditEvent event = createEvent(entity, EntityAuditAction.TAG_ADD,
                                                      "Added trait: " + InstanceSerialization.toJson(trait, true));

                 auditRepository.putEvents(event);
             }
         }
     }

     @Override
     public void onTraitsDeleted(ITypedReferenceableInstance entity, Collection<String> traitNames) throws AtlasException {
         if (traitNames != null) {
             for (String traitName : traitNames) {
                 EntityAuditEvent event = createEvent(entity, EntityAuditAction.TAG_DELETE, "Deleted trait: " + traitName);

                 auditRepository.putEvents(event);
             }
         }
     }

     @Override
     public void onTraitsUpdated(ITypedReferenceableInstance entity, Collection<? extends IStruct> traits) throws AtlasException {
         if (traits != null) {
             for (IStruct trait : traits) {
                 EntityAuditEvent event = createEvent(entity, EntityAuditAction.TAG_UPDATE,
                                                      "Updated trait: " + InstanceSerialization.toJson(trait, true));

                 auditRepository.putEvents(event);
             }
         }
     }

     @Override
     public void onEntitiesDeleted(Collection<ITypedReferenceableInstance> entities, boolean isImport) throws AtlasException {
         List<EntityAuditEvent> events = new ArrayList<>();
         for (ITypedReferenceableInstance entity : entities) {
             EntityAuditEvent event = createEvent(entity, isImport ? EntityAuditAction.ENTITY_IMPORT_DELETE : EntityAuditAction.ENTITY_DELETE, "Deleted entity");
             events.add(event);
         }

         auditRepository.putEvents(events);
     }

     public List<EntityAuditEvent> getAuditEvents(String guid) throws AtlasException{
         return auditRepository.listEvents(guid, null, (short) 10);
     }

     private EntityAuditEvent createEvent(ITypedReferenceableInstance entity, EntityAuditAction action)
             throws AtlasException {
         String detail = getAuditEventDetail(entity, action);

         return createEvent(entity, action, detail);
     }

     private EntityAuditEvent createEvent(ITypedReferenceableInstance entity, EntityAuditAction action, String details)
             throws AtlasException {
         return new EntityAuditEvent(entity.getId()._getId(), RequestContextV1.get().getRequestTime(), RequestContextV1.get().getUser(), action, details, entity);
     }

     private String getAuditEventDetail(ITypedReferenceableInstance entity, EntityAuditAction action) throws AtlasException {
         Map<String, Object> prunedAttributes = pruneEntityAttributesForAudit(entity);

         String auditPrefix  = getAuditPrefix(action);
         String auditString  = auditPrefix + InstanceSerialization.toJson(entity, true);
         byte[] auditBytes   = auditString.getBytes(StandardCharsets.UTF_8);
         long   auditSize    = auditBytes != null ? auditBytes.length : 0;
         long   auditMaxSize = auditRepository.repositoryMaxSize();

         if (auditMaxSize >= 0 && auditSize > auditMaxSize) { // don't store attributes in audit
             LOG.warn("audit record too long: entityType={}, guid={}, size={}; maxSize={}. entity attribute values not stored in audit",
                     entity.getTypeName(), entity.getId()._getId(), auditSize, auditMaxSize);

             Map<String, Object> attrValues = entity.getValuesMap();

             clearAttributeValues(entity);

             auditString = auditPrefix + InstanceSerialization.toJson(entity, true);

             addAttributeValues(entity, attrValues);
         }

         restoreEntityAttributes(entity, prunedAttributes);

         return auditString;
     }

     private void clearAttributeValues(IReferenceableInstance entity) throws AtlasException {
         Map<String, Object> attributesMap = entity.getValuesMap();

         if (MapUtils.isNotEmpty(attributesMap)) {
             for (String attribute : attributesMap.keySet()) {
                 entity.setNull(attribute);
             }
         }
     }

     private void addAttributeValues(ITypedReferenceableInstance entity, Map<String, Object> attributesMap) throws AtlasException {
         if (MapUtils.isNotEmpty(attributesMap)) {
             for (String attr : attributesMap.keySet()) {
                 entity.set(attr, attributesMap.get(attr));
             }
         }
     }

     private Map<String, Object> pruneEntityAttributesForAudit(ITypedReferenceableInstance entity) throws AtlasException {
         Map<String, Object> ret               = null;
         Map<String, Object> entityAttributes  = entity.getValuesMap();
         List<String>        excludeAttributes = auditRepository.getAuditExcludeAttributes(entity.getTypeName());

         if (CollectionUtils.isNotEmpty(excludeAttributes) && MapUtils.isNotEmpty(entityAttributes)) {
             Map<String, AttributeInfo> attributeInfoMap = entity.fieldMapping().fields;

             for (String attrName : entityAttributes.keySet()) {
                 Object        attrValue = entityAttributes.get(attrName);
                 AttributeInfo attrInfo  = attributeInfoMap.get(attrName);

                 if (excludeAttributes.contains(attrName)) {
                     if (ret == null) {
                         ret = new HashMap<>();
                     }

                     ret.put(attrName, attrValue);
                     entity.setNull(attrName);
                 } else if (attrInfo.isComposite) {
                     if (attrValue instanceof Collection) {
                         for (Object attribute : (Collection) attrValue) {
                             if (attribute instanceof ITypedReferenceableInstance) {
                                 ret = pruneAttributes(ret, (ITypedReferenceableInstance) attribute);
                             }
                         }
                     } else if (attrValue instanceof ITypedReferenceableInstance) {
                         ret = pruneAttributes(ret, (ITypedReferenceableInstance) attrValue);
                     }
                 }
             }
         }

         return ret;
     }

     private Map<String, Object> pruneAttributes(Map<String, Object> ret, ITypedReferenceableInstance attribute) throws AtlasException {
         ITypedReferenceableInstance attrInstance = attribute;
         Map<String, Object>         prunedAttrs  = pruneEntityAttributesForAudit(attrInstance);

         if (MapUtils.isNotEmpty(prunedAttrs)) {
             if (ret == null) {
                 ret = new HashMap<>();
             }

             ret.put(attrInstance.getId()._getId(), prunedAttrs);
         }
         return ret;
     }

     private void restoreEntityAttributes(ITypedReferenceableInstance entity, Map<String, Object> prunedAttributes) throws AtlasException {
         if (MapUtils.isEmpty(prunedAttributes)) {
             return;
         }

         Map<String, Object> entityAttributes = entity.getValuesMap();

         if (MapUtils.isNotEmpty(entityAttributes)) {
             Map<String, AttributeInfo> attributeInfoMap = entity.fieldMapping().fields;

             for (String attrName : entityAttributes.keySet()) {
                 Object        attrValue = entityAttributes.get(attrName);
                 AttributeInfo attrInfo  = attributeInfoMap.get(attrName);

                 if (prunedAttributes.containsKey(attrName)) {
                     entity.set(attrName, prunedAttributes.get(attrName));
                 } else if (attrInfo.isComposite) {
                     if (attrValue instanceof Collection) {
                         for (Object attributeEntity : (Collection) attrValue) {
                             if (attributeEntity instanceof ITypedReferenceableInstance) {
                                 restoreAttributes(prunedAttributes, (ITypedReferenceableInstance) attributeEntity);
                             }
                         }
                     } else if (attrValue instanceof ITypedReferenceableInstance) {
                         restoreAttributes(prunedAttributes, (ITypedReferenceableInstance) attrValue);
                     }
                 }
             }
         }
     }

     private void restoreAttributes(Map<String, Object> prunedAttributes, ITypedReferenceableInstance attributeEntity) throws AtlasException {
         Object                      obj          = prunedAttributes.get(attributeEntity.getId()._getId());

         if (obj instanceof Map) {
             restoreEntityAttributes(attributeEntity, (Map) obj);
         }
     }

     private String getAuditPrefix(EntityAuditAction action) {
         final String ret;

         switch (action) {
             case ENTITY_CREATE:
                 ret = "Created: ";
                 break;
             case ENTITY_UPDATE:
                 ret = "Updated: ";
                 break;
             case ENTITY_DELETE:
                 ret = "Deleted: ";
                 break;
             case TAG_ADD:
                 ret = "Added trait: ";
                 break;
             case TAG_DELETE:
                 ret = "Deleted trait: ";
                 break;
             case TAG_UPDATE:
                 ret = "Updated trait: ";
                 break;
             case ENTITY_IMPORT_CREATE:
                 ret = "Created by import: ";
                 break;
             case ENTITY_IMPORT_UPDATE:
                 ret = "Updated by import: ";
                 break;
             case ENTITY_IMPORT_DELETE:
                 ret = "Deleted by import: ";
                 break;
             default:
                 ret = "Unknown: ";
         }

         return ret;
     }
 }
	/**
	* 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.repository.audit;

	import org.apache.atlas.AtlasException;
	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.typesystem.IReferenceableInstance;
	import org.apache.atlas.typesystem.IStruct;
	import org.apache.atlas.typesystem.ITypedReferenceableInstance;
	import org.apache.atlas.typesystem.json.InstanceSerialization;
	import org.apache.atlas.typesystem.types.AttributeInfo;
	import org.apache.commons.collections.CollectionUtils;
	import org.apache.commons.collections.MapUtils;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import javax.inject.Inject;
	import java.nio.charset.StandardCharsets;
	import java.util.ArrayList;
	import java.util.Collection;
	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;

	/**
	* Listener on entity create/update/delete, tag add/delete. Adds the corresponding audit event to the audit repository.
	*/
	public class EntityAuditListener implements EntityChangeListener {
	private static final Logger LOG = LoggerFactory.getLogger(EntityAuditListener.class);

	private EntityAuditRepository auditRepository;

	@Inject
	public EntityAuditListener(EntityAuditRepository auditRepository) {
	this.auditRepository = auditRepository;
	}

	@Override
	public void onEntitiesAdded(Collection<ITypedReferenceableInstance> entities, boolean isImport) throws AtlasException {
	List<EntityAuditEvent> events = new ArrayList<>();
	for (ITypedReferenceableInstance entity : entities) {
	EntityAuditEvent event = createEvent(entity, isImport ? EntityAuditAction.ENTITY_IMPORT_CREATE : EntityAuditAction.ENTITY_CREATE);
	events.add(event);
	}

	auditRepository.putEvents(events);
	}

	@Override
	public void onEntitiesUpdated(Collection<ITypedReferenceableInstance> entities, boolean isImport) throws AtlasException {
	List<EntityAuditEvent> events = new ArrayList<>();
	for (ITypedReferenceableInstance entity : entities) {
	EntityAuditEvent event = createEvent(entity, isImport ? EntityAuditAction.ENTITY_IMPORT_UPDATE : EntityAuditAction.ENTITY_UPDATE);
	events.add(event);
	}

	auditRepository.putEvents(events);
	}

	@Override
	public void onTraitsAdded(ITypedReferenceableInstance entity, Collection<? extends IStruct> traits) throws AtlasException {
	if (traits != null) {
	for (IStruct trait : traits) {
	EntityAuditEvent event = createEvent(entity, EntityAuditAction.TAG_ADD,
	"Added trait: " + InstanceSerialization.toJson(trait, true));

	auditRepository.putEvents(event);
	}
	}
	}

	@Override
	public void onTraitsDeleted(ITypedReferenceableInstance entity, Collection<String> traitNames) throws AtlasException {
	if (traitNames != null) {
	for (String traitName : traitNames) {
	EntityAuditEvent event = createEvent(entity, EntityAuditAction.TAG_DELETE, "Deleted trait: " + traitName);

	auditRepository.putEvents(event);
	}
	}
	}

	@Override
	public void onTraitsUpdated(ITypedReferenceableInstance entity, Collection<? extends IStruct> traits) throws AtlasException {
	if (traits != null) {
	for (IStruct trait : traits) {
	EntityAuditEvent event = createEvent(entity, EntityAuditAction.TAG_UPDATE,
	"Updated trait: " + InstanceSerialization.toJson(trait, true));

	auditRepository.putEvents(event);
	}
	}
	}

	@Override
	public void onEntitiesDeleted(Collection<ITypedReferenceableInstance> entities, boolean isImport) throws AtlasException {
	List<EntityAuditEvent> events = new ArrayList<>();
	for (ITypedReferenceableInstance entity : entities) {
	EntityAuditEvent event = createEvent(entity, isImport ? EntityAuditAction.ENTITY_IMPORT_DELETE : EntityAuditAction.ENTITY_DELETE, "Deleted entity");
	events.add(event);
	}

	auditRepository.putEvents(events);
	}

	public List<EntityAuditEvent> getAuditEvents(String guid) throws AtlasException{
	return auditRepository.listEvents(guid, null, (short) 10);
	}

	private EntityAuditEvent createEvent(ITypedReferenceableInstance entity, EntityAuditAction action)
	throws AtlasException {
	String detail = getAuditEventDetail(entity, action);

	return createEvent(entity, action, detail);
	}

	private EntityAuditEvent createEvent(ITypedReferenceableInstance entity, EntityAuditAction action, String details)
	throws AtlasException {
	return new EntityAuditEvent(entity.getId()._getId(), RequestContextV1.get().getRequestTime(), RequestContextV1.get().getUser(), action, details, entity);
	}

	private String getAuditEventDetail(ITypedReferenceableInstance entity, EntityAuditAction action) throws AtlasException {
	Map<String, Object> prunedAttributes = pruneEntityAttributesForAudit(entity);

	String auditPrefix = getAuditPrefix(action);
	String auditString = auditPrefix + InstanceSerialization.toJson(entity, true);
	byte[] auditBytes = auditString.getBytes(StandardCharsets.UTF_8);
	long auditSize = auditBytes != null ? auditBytes.length : 0;
	long auditMaxSize = auditRepository.repositoryMaxSize();

	if (auditMaxSize >= 0 && auditSize > auditMaxSize) { // don't store attributes in audit
	LOG.warn("audit record too long: entityType={}, guid={}, size={}; maxSize={}. entity attribute values not stored in audit",
	entity.getTypeName(), entity.getId()._getId(), auditSize, auditMaxSize);

	Map<String, Object> attrValues = entity.getValuesMap();

	clearAttributeValues(entity);

	auditString = auditPrefix + InstanceSerialization.toJson(entity, true);

	addAttributeValues(entity, attrValues);
	}

	restoreEntityAttributes(entity, prunedAttributes);

	return auditString;
	}

	private void clearAttributeValues(IReferenceableInstance entity) throws AtlasException {
	Map<String, Object> attributesMap = entity.getValuesMap();

	if (MapUtils.isNotEmpty(attributesMap)) {
	for (String attribute : attributesMap.keySet()) {
	entity.setNull(attribute);
	}
	}
	}

	private void addAttributeValues(ITypedReferenceableInstance entity, Map<String, Object> attributesMap) throws AtlasException {
	if (MapUtils.isNotEmpty(attributesMap)) {
	for (String attr : attributesMap.keySet()) {
	entity.set(attr, attributesMap.get(attr));
	}
	}
	}

	private Map<String, Object> pruneEntityAttributesForAudit(ITypedReferenceableInstance entity) throws AtlasException {
	Map<String, Object> ret = null;
	Map<String, Object> entityAttributes = entity.getValuesMap();
	List<String> excludeAttributes = auditRepository.getAuditExcludeAttributes(entity.getTypeName());

	if (CollectionUtils.isNotEmpty(excludeAttributes) && MapUtils.isNotEmpty(entityAttributes)) {
	Map<String, AttributeInfo> attributeInfoMap = entity.fieldMapping().fields;

	for (String attrName : entityAttributes.keySet()) {
	Object attrValue = entityAttributes.get(attrName);
	AttributeInfo attrInfo = attributeInfoMap.get(attrName);

	if (excludeAttributes.contains(attrName)) {
	if (ret == null) {
	ret = new HashMap<>();
	}

	ret.put(attrName, attrValue);
	entity.setNull(attrName);
	} else if (attrInfo.isComposite) {
	if (attrValue instanceof Collection) {
	for (Object attribute : (Collection) attrValue) {
	if (attribute instanceof ITypedReferenceableInstance) {
	ret = pruneAttributes(ret, (ITypedReferenceableInstance) attribute);
	}
	}
	} else if (attrValue instanceof ITypedReferenceableInstance) {
	ret = pruneAttributes(ret, (ITypedReferenceableInstance) attrValue);
	}
	}
	}
	}

	return ret;
	}

	private Map<String, Object> pruneAttributes(Map<String, Object> ret, ITypedReferenceableInstance attribute) throws AtlasException {
	ITypedReferenceableInstance attrInstance = attribute;
	Map<String, Object> prunedAttrs = pruneEntityAttributesForAudit(attrInstance);

	if (MapUtils.isNotEmpty(prunedAttrs)) {
	if (ret == null) {
	ret = new HashMap<>();
	}

	ret.put(attrInstance.getId()._getId(), prunedAttrs);
	}
	return ret;
	}

	private void restoreEntityAttributes(ITypedReferenceableInstance entity, Map<String, Object> prunedAttributes) throws AtlasException {
	if (MapUtils.isEmpty(prunedAttributes)) {
	return;
	}

	Map<String, Object> entityAttributes = entity.getValuesMap();

	if (MapUtils.isNotEmpty(entityAttributes)) {
	Map<String, AttributeInfo> attributeInfoMap = entity.fieldMapping().fields;

	for (String attrName : entityAttributes.keySet()) {
	Object attrValue = entityAttributes.get(attrName);
	AttributeInfo attrInfo = attributeInfoMap.get(attrName);

	if (prunedAttributes.containsKey(attrName)) {
	entity.set(attrName, prunedAttributes.get(attrName));
	} else if (attrInfo.isComposite) {
	if (attrValue instanceof Collection) {
	for (Object attributeEntity : (Collection) attrValue) {
	if (attributeEntity instanceof ITypedReferenceableInstance) {
	restoreAttributes(prunedAttributes, (ITypedReferenceableInstance) attributeEntity);
	}
	}
	} else if (attrValue instanceof ITypedReferenceableInstance) {
	restoreAttributes(prunedAttributes, (ITypedReferenceableInstance) attrValue);
	}
	}
	}
	}
	}

	private void restoreAttributes(Map<String, Object> prunedAttributes, ITypedReferenceableInstance attributeEntity) throws AtlasException {
	Object obj = prunedAttributes.get(attributeEntity.getId()._getId());

	if (obj instanceof Map) {
	restoreEntityAttributes(attributeEntity, (Map) obj);
	}
	}

	private String getAuditPrefix(EntityAuditAction action) {
	final String ret;

	switch (action) {
	case ENTITY_CREATE:
	ret = "Created: ";
	break;
	case ENTITY_UPDATE:
	ret = "Updated: ";
	break;
	case ENTITY_DELETE:
	ret = "Deleted: ";
	break;
	case TAG_ADD:
	ret = "Added trait: ";
	break;
	case TAG_DELETE:
	ret = "Deleted trait: ";
	break;
	case TAG_UPDATE:
	ret = "Updated trait: ";
	break;
	case ENTITY_IMPORT_CREATE:
	ret = "Created by import: ";
	break;
	case ENTITY_IMPORT_UPDATE:
	ret = "Updated by import: ";
	break;
	case ENTITY_IMPORT_DELETE:
	ret = "Deleted by import: ";
	break;
	default:
	ret = "Unknown: ";
	}

	return ret;
	}
	}