ATLAS-5187: Fix audit related issues when rdbms backend is selected

graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/dao/DbEntityAuditDao.java

@@ -18,6 +18,7 @@
 package org.apache.atlas.repository.audit.rdbms.dao;
 
 import org.apache.atlas.repository.audit.rdbms.entity.DbEntityAudit;
+import org.apache.commons.collections.CollectionUtils;
 
 import javax.persistence.EntityManager;
 import javax.persistence.NoResultException;
@@ -30,34 +31,52 @@ public DbEntityAuditDao(EntityManager em) {
         super(em);
     }
 
-    public List<DbEntityAudit> getByEntityIdActionStartTimeStartIdx(String entityId, int action, long eventTimeStart, int eventIdxStart, int maxResults) {
+    public List<DbEntityAudit> getByEntityIdActionStartTimeStartIdx(String entityId, String action, long eventTimeStart, int eventIdxStart, int maxResults) {
         try {
-            return em.createNamedQuery("DbEntityAudit.getByEntityIdActionStartTimeStartIdx", DbEntityAudit.class)
-                    .setParameter("entityId", entityId)
-                    .setParameter("action", action)
-                    .setParameter("eventTimeStart", eventTimeStart)
-                    .setParameter("eventIdxStart", eventIdxStart)
-                    .setMaxResults(maxResults)
-                    .getResultList();
+            if (action == null) {
+                return em.createNamedQuery("DbEntityAudit.getByEntityIdStartTimeStartIdx", DbEntityAudit.class)
+                        .setParameter("entityId", entityId)
+                        .setParameter("eventTimeStart", eventTimeStart)
+                        .setParameter("eventIdxStart", eventIdxStart)
+                        .setMaxResults(maxResults)
+                        .getResultList();
+            } else {
+                return em.createNamedQuery("DbEntityAudit.getByEntityIdActionStartTimeStartIdx", DbEntityAudit.class)
+                        .setParameter("entityId", entityId)
+                        .setParameter("action", action)
+                        .setParameter("eventTimeStart", eventTimeStart)
+                        .setParameter("eventIdxStart", eventIdxStart)
+                        .setMaxResults(maxResults)
+                        .getResultList();
+            }
         } catch (NoResultException excp) {
             // ignore
         }
 
         return Collections.emptyList();
     }
 
-    public List<DbEntityAudit> getByEntityIdAction(String entityId, Integer action, int startIdx, int maxResults) {
+    public List<DbEntityAudit> getByEntityIdAction(String entityId, String action, List<String> sortByColumn, boolean sortOrder, int startIdx, int maxResults) {
         try {
-            if (action == null) {
-                return em.createNamedQuery("DbEntityAudit.getByEntityId", DbEntityAudit.class)
+            StringBuilder query = new StringBuilder("SELECT e FROM DbEntityAudit e WHERE e.entityId = :entityId");
+
+            if (action != null) {
+                query.append(" and e.action = :action");
+                if (CollectionUtils.isNotEmpty(sortByColumn)) {
+                    query.append(getOrderByQuery(sortByColumn, sortOrder));
+                }
+                return em.createQuery(query.toString(), DbEntityAudit.class)
                         .setParameter("entityId", entityId)
+                        .setParameter("action", action)
                         .setFirstResult(startIdx)
                         .setMaxResults(maxResults)
                         .getResultList();
             } else {
-                return em.createNamedQuery("DbEntityAudit.getByEntityIdAction", DbEntityAudit.class)
+                if (CollectionUtils.isNotEmpty(sortByColumn)) {
+                    query.append(getOrderByQuery(sortByColumn, sortOrder));
+                }
+                return em.createQuery(query.toString(), DbEntityAudit.class)
                         .setParameter("entityId", entityId)
-                        .setParameter("action", action)
                         .setFirstResult(startIdx)
                         .setMaxResults(maxResults)
                         .getResultList();
@@ -68,4 +87,50 @@ public List<DbEntityAudit> getByEntityIdAction(String entityId, Integer action,
 
         return Collections.emptyList();
     }
+
+    public String getOrderByQuery(List<String> sortByColumn, boolean sortOrderDesc) {
+        StringBuilder orderByQuery = new StringBuilder(" ORDER BY ");
+        for (int i = 0; i < sortByColumn.size(); i++) {
+            orderByQuery.append("e.").append(sortByColumn.get(i));
+            orderByQuery.append(sortOrderDesc ? " DESC" : " ASC");
+            if (i != sortByColumn.size() - 1) {
+                orderByQuery.append(", ");
+            }
+        }
+        return orderByQuery.toString();
+    }
+
+    public List<DbEntityAudit> getLatestAuditsByEntityIdAction(String entityId, String action, List<String> filterActions) {
+        try {
+            StringBuilder query = new StringBuilder("SELECT e FROM DbEntityAudit e WHERE e.entityId = :entityId");
+
+            if (action != null) {
+                query.append(" and e.action = :action");
+                query.append(" ORDER BY e.eventTime DESC, e.eventIndex DESC");
+                return em.createQuery(query.toString(), DbEntityAudit.class)
+                        .setParameter("entityId", entityId)
+                        .setParameter("action", action)
+                        .getResultList();
+            } else {
+                if (CollectionUtils.isNotEmpty(filterActions)) {
+                    query.append(" and e.action NOT IN (");
+                    for (int i = 0; i < filterActions.size(); i++) {
+                        query.append("'").append(filterActions.get(i)).append("'");
+                        if (i != filterActions.size() - 1) {
+                            query.append(", ");
+                        }
+                    }
+                    query.append(")");
+                }
+                query.append(" ORDER BY e.eventTime DESC, e.eventIndex DESC");
+                return em.createQuery(query.toString(), DbEntityAudit.class)
+                        .setParameter("entityId", entityId)
+                        .getResultList();
+            }
+        } catch (NoResultException excp) {
+            // ignore
+        }
+
+        return Collections.emptyList();
+    }
 }

graphdb/janusgraph-rdbms/src/main/java/org/apache/atlas/repository/audit/rdbms/entity/DbEntityAudit.java

@@ -61,8 +61,8 @@ public class DbEntityAudit implements java.io.Serializable {
     @Column(name = "user_name", nullable = false, length = 64)
     protected String user;
 
-    @Column(name = "operation", nullable = false)
-    protected int action;
+    @Column(name = "operation", nullable = false, length = 64)
+    protected String action;
 
     @Column(name = "details")
     @Lob
@@ -115,11 +115,11 @@ public void setUser(String user) {
         this.user = user;
     }
 
-    public int getAction() {
+    public String getAction() {
         return action;
     }
 
-    public void setAction(int action) {
+    public void setAction(String action) {
         this.action = action;
     }
 
@@ -164,7 +164,7 @@ public boolean equals(Object obj) {
                     eventTime == other.eventTime &&
                     eventIndex == other.eventIndex &&
                     Objects.equals(user, other.user) &&
-                    action == other.action &&
+                    Objects.equals(action, other.action) &&
                     Objects.equals(details, other.details) &&
                     Objects.equals(entity, other.entity) &&
                     auditType == other.auditType;

graphdb/janusgraph-rdbms/src/main/resources/META-INF/janus-jpa_named_queries.xml

@@ -13,25 +13,19 @@
                  xmlns="http://java.sun.com/xml/ns/persistence/orm" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
                  xsi:schemaLocation="http://java.sun.com/xml/ns/persistence/orm http://java.sun.com/xml/ns/persistence/orm_1_0.xsd ">
 
-    <named-query name="DbEntityAudit.getByEntityId">
-        <query>SELECT obj FROM DbEntityAudit obj
-            WHERE obj.entityId = :entityId
-            ORDER BY obj.eventTime DESC, obj.eventIndex DESC
-        </query>
-    </named-query>
-
-    <named-query name="DbEntityAudit.getByEntityIdAction">
+    <named-query name="DbEntityAudit.getByEntityIdActionStartTimeStartIdx">
         <query>SELECT obj FROM DbEntityAudit obj
             WHERE obj.entityId = :entityId
             AND obj.action     = :action
+            AND obj.eventTime  >= :eventTimeStart
+            AND obj.eventIndex >= :eventIdxStart
             ORDER BY obj.eventTime DESC, obj.eventIndex DESC
         </query>
     </named-query>
 
-    <named-query name="DbEntityAudit.getByEntityIdActionStartTimeStartIdx">
+    <named-query name="DbEntityAudit.getByEntityIdStartTimeStartIdx">
         <query>SELECT obj FROM DbEntityAudit obj
             WHERE obj.entityId = :entityId
-            AND obj.action     = :action
             AND obj.eventTime  >= :eventTimeStart
             AND obj.eventIndex >= :eventIdxStart
             ORDER BY obj.eventTime DESC, obj.eventIndex DESC

graphdb/janusgraph-rdbms/src/main/resources/META-INF/postgres/create_schema.sql

@@ -16,7 +16,7 @@
 -- DB objects for Atlas entity audit
 CREATE SEQUENCE IF NOT EXISTS atlas_entity_audit_seq INCREMENT BY 1 CACHE 1000;
 
-CREATE TABLE IF NOT EXISTS atlas_entity_audit(id BIGINT, entity_id VARCHAR(64) NOT NULL, event_time BIGINT NOT NULL, event_idx INT NOT NULL, user_name  VARCHAR(64) NOT NULL, operation INT NOT NULL, details TEXT DEFAULT NULL, entity TEXT DEFAULT NULL, audit_type INT NOT NULL, PRIMARY KEY(id));
+CREATE TABLE IF NOT EXISTS atlas_entity_audit(id BIGINT, entity_id VARCHAR(64) NOT NULL, event_time BIGINT NOT NULL, event_idx INT NOT NULL, user_name  VARCHAR(64) NOT NULL, operation VARCHAR(64) NOT NULL, details TEXT DEFAULT NULL, entity TEXT DEFAULT NULL, audit_type INT NOT NULL, PRIMARY KEY(id));
 
 CREATE INDEX IF NOT EXISTS atlas_entity_audit_idx_entity_id            ON atlas_entity_audit (entity_id);
 CREATE INDEX IF NOT EXISTS atlas_entity_audit_idx_event_time           ON atlas_entity_audit (event_time);

repository/src/main/java/org/apache/atlas/repository/audit/AbstractStorageBasedAuditRepository.java

@@ -24,10 +24,10 @@
 import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.listener.ActiveStateChangeHandler;
 import org.apache.atlas.model.audit.EntityAuditEventV2;
-import org.apache.atlas.repository.Constants.AtlasAuditAgingType;
 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.util.Bytes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -36,7 +36,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 /**
  * This abstract base class should be used when adding support for an audit storage backend.
@@ -83,11 +82,6 @@ public void putEventsV2(EntityAuditEventV2... events) throws AtlasBaseException
         putEventsV2(Arrays.asList(events));
     }
 
-    @Override
-    public List<EntityAuditEventV2> deleteEventsV2(String entityId, Set<EntityAuditEventV2.EntityAuditActionV2> entityAuditActions, short auditCount, int ttlInDays, boolean createEventsAgeoutAllowed, AtlasAuditAgingType auditAgingType) throws AtlasBaseException, AtlasException {
-        return null;
-    }
-
     @Override
     public List<Object> listEvents(String entityId, String startKey, short maxResults) throws AtlasBaseException {
         List ret = listEventsV2(entityId, null, startKey, maxResults);
@@ -168,6 +162,10 @@ protected byte[] getKey(String id, Long ts, int index) {
     }
 
     protected long getTimestampFromKey(String key) {
+        if (StringUtils.isEmpty(key)) {
+            return 0L;
+        }
+
         String[] parts = key.split(FIELD_SEPARATOR);
 
         if (parts.length < 3) {
@@ -183,6 +181,10 @@ protected long getTimestampFromKey(String key) {
     }
 
     protected int getIndexFromKey(String key) {
+        if (StringUtils.isEmpty(key)) {
+            return 0;
+        }
+
         String[] parts = key.split(FIELD_SEPARATOR);
 
         if (parts.length < 3) {

repository/src/main/java/org/apache/atlas/repository/audit/CassandraBasedAuditRepository.java

@@ -28,7 +28,9 @@
 import org.apache.atlas.AtlasException;
 import org.apache.atlas.EntityAuditEvent;
 import org.apache.atlas.annotation.ConditionalOnAtlasProperty;
+import org.apache.atlas.exception.AtlasBaseException;
 import org.apache.atlas.model.audit.EntityAuditEventV2;
+import org.apache.atlas.repository.Constants;
 import org.apache.commons.lang.NotImplementedException;
 import org.apache.commons.lang.StringUtils;
 import org.slf4j.Logger;
@@ -38,6 +40,7 @@
 import javax.inject.Singleton;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -203,6 +206,13 @@ public List<EntityAuditEventV2> listEventsV2(String entityId, EntityAuditEventV2
         return listEventsV2(entityId, auditAction, null, limit);
     }
 
+    @Override
+    public List<EntityAuditEventV2> deleteEventsV2(String entityId, Set<EntityAuditEventV2.EntityAuditActionV2> entityAuditActions, short auditCount, int ttlInDays, boolean createEventsAgeoutAllowed, Constants.AtlasAuditAgingType auditAgingType)
+            throws AtlasBaseException, AtlasException
+    {
+        return Collections.emptyList();
+    }
+
     @Override
     public Set<String> getEntitiesWithTagChanges(long fromTimestamp, long toTimestamp) {
         throw new NotImplementedException();