RANGER-5482:Create Ranger Audit Server with SOLR and HDFS as audit consumer

.gitignore

@@ -19,3 +19,9 @@ winpkg/target
 .python-version
 /security-admin/src/main/webapp/react-webapp/node_modules
 **/target
+
+# Runtime logs and process files
+logs/
+*.log
+*.pid
+catalina.out

agents-audit/core/pom.xml

@@ -63,6 +63,11 @@
                 </exclusion>
             </exclusions>
         </dependency>
+        <dependency>
+            <groupId>javax.xml.bind</groupId>
+            <artifactId>jaxb-api</artifactId>
+            <version>${jaxb.api.version}</version>
+        </dependency>
         <dependency>
             <groupId>joda-time</groupId>
             <artifactId>joda-time</artifactId>

agents-audit/core/src/main/java/org/apache/ranger/audit/model/AuthzAuditEvent.java

@@ -19,16 +19,28 @@
 
 package org.apache.ranger.audit.model;
 
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import org.apache.commons.lang3.StringUtils;
 
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlRootElement;
+
 import java.util.Date;
 import java.util.HashSet;
 import java.util.Set;
 
+@JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.ANY)
+@JsonInclude(JsonInclude.Include.NON_EMPTY)
+@JsonIgnoreProperties(ignoreUnknown = true)
 @JsonSerialize
+@XmlRootElement
+@XmlAccessorType(XmlAccessType.FIELD)
 public class AuthzAuditEvent extends AuditEventBase {
     protected static final int    MAX_ACTION_FIELD_SIZE       = 1800;
     protected static final int    MAX_REQUEST_DATA_FIELD_SIZE = 1800;

agents-audit/core/src/main/java/org/apache/ranger/audit/provider/AuditHandler.java

@@ -29,6 +29,8 @@ public interface AuditHandler {
 
     boolean log(Collection<AuditEventBase> events);
 
+    boolean log(Collection<AuditEventBase> events, String batchKey);
+
     boolean logJSON(String event);
 
     boolean logJSON(Collection<String> events);

agents-audit/core/src/main/java/org/apache/ranger/audit/provider/AuditProviderFactory.java

@@ -439,6 +439,8 @@ private AuditHandler getProviderFromConfig(Properties props, String propPrefix,
                 provider = createDestination("org.apache.ranger.audit.provider.kafka.KafkaAuditProvider");
             } else if (providerName.equalsIgnoreCase("log4j")) {
                 provider = createDestination("org.apache.ranger.audit.destination.Log4JAuditDestination");
+            } else if (providerName.equalsIgnoreCase("auditserver")) {
+                provider = createDestination("org.apache.ranger.audit.destination.RangerAuditServerDestination");
             } else if (providerName.equalsIgnoreCase("batch")) {
                 provider = getAuditProvider(props, propPrefix, consumer);
             } else if (providerName.equalsIgnoreCase("async")) {

agents-audit/core/src/main/java/org/apache/ranger/audit/provider/BaseAuditHandler.java

@@ -72,18 +72,19 @@ public abstract class BaseAuditHandler implements AuditHandler {
     protected Map<String, String> configProps      = new HashMap<>();
     protected Properties          props;
 
-    int     errorLogIntervalMS = 30 * 1000; // Every 30 seconds
-    long    lastErrorLogMS;
-    long    totalCount;
-    long    totalSuccessCount;
-    long    totalFailedCount;
-    long    totalStashedCount;
-    long    totalDeferredCount;
-    long    lastIntervalCount;
-    long    lastIntervalSuccessCount;
-    long    lastIntervalFailedCount;
-    long    lastStashedCount;
-    long    lastDeferredCount;
+    int        errorLogIntervalMS      = 30 * 1000; // Every 30 seconds
+    long       lastErrorLogMS;
+    long       lastIntervalCount;
+    long       lastIntervalSuccessCount;
+    long       lastIntervalFailedCount;
+    long       lastStashedCount;
+    long       lastDeferredCount;
+    AtomicLong totalCount              = new AtomicLong(0);
+    AtomicLong totalSuccessCount       = new AtomicLong(0);
+    AtomicLong totalFailedCount        = new AtomicLong(0);
+    AtomicLong totalStashedCount       = new AtomicLong(0);
+    AtomicLong totalDeferredCount      = new AtomicLong(0);
+
     boolean statusLogEnabled    = DEFAULT_AUDIT_LOG_STATUS_LOG_ENABLED;
     long    statusLogIntervalMS = DEFAULT_AUDIT_LOG_STATUS_LOG_INTERVAL_SEC * 1000;
     long    lastStatusLogTime   = System.currentTimeMillis();
@@ -106,6 +107,11 @@ public boolean log(AuditEventBase event) {
         return log(Collections.singletonList(event));
     }
 
+    @Override
+    public boolean log(Collection<AuditEventBase> events, String batchKey) {
+        return log(events);
+    }
+
     /*
      * (non-Javadoc)
      *
@@ -237,57 +243,47 @@ public String getFinalPath() {
     }
 
     public long addTotalCount(int count) {
-        totalCount += count;
-
-        return totalCount;
+        return totalCount.addAndGet(count);
     }
 
     public long addSuccessCount(int count) {
-        totalSuccessCount += count;
-
-        return totalSuccessCount;
+        return totalSuccessCount.addAndGet(count);
     }
 
     public long addFailedCount(int count) {
-        totalFailedCount += count;
-
-        return totalFailedCount;
+        return totalFailedCount.addAndGet(count);
     }
 
     public long addStashedCount(int count) {
-        totalStashedCount += count;
-
-        return totalStashedCount;
+        return totalStashedCount.addAndGet(count);
     }
 
     public long addDeferredCount(int count) {
-        totalDeferredCount += count;
-
-        return totalDeferredCount;
+        return totalDeferredCount.addAndGet(count);
     }
 
     public long getTotalCount() {
-        return totalCount;
+        return totalCount.get();
     }
 
     public long getTotalSuccessCount() {
-        return totalSuccessCount;
+        return totalSuccessCount.get();
     }
 
     public long getTotalFailedCount() {
-        return totalFailedCount;
+        return totalFailedCount.get();
     }
 
     public long getTotalStashedCount() {
-        return totalStashedCount;
+        return totalStashedCount.get();
     }
 
     public long getLastStashedCount() {
         return lastStashedCount;
     }
 
     public long getTotalDeferredCount() {
-        return totalDeferredCount;
+        return totalDeferredCount.get();
     }
 
     public long getLastDeferredCount() {
@@ -312,21 +308,27 @@ public void logStatus() {
             lastStatusLogTime = currTime;
             nextStatusLogTime = currTime + statusLogIntervalMS;
 
-            long diffCount    = totalCount - lastIntervalCount;
-            long diffSuccess  = totalSuccessCount - lastIntervalSuccessCount;
-            long diffFailed   = totalFailedCount - lastIntervalFailedCount;
-            long diffStashed  = totalStashedCount - lastStashedCount;
-            long diffDeferred = totalDeferredCount - lastDeferredCount;
+            long currentTotalCount   = totalCount.get();
+            long currentSuccessCount = totalSuccessCount.get();
+            long currentFailedCount  = totalFailedCount.get();
+            long currentStashedCount = totalStashedCount.get();
+            long currentDeferredCount = totalDeferredCount.get();
+
+            long diffCount    = currentTotalCount - lastIntervalCount;
+            long diffSuccess  = currentSuccessCount - lastIntervalSuccessCount;
+            long diffFailed   = currentFailedCount - lastIntervalFailedCount;
+            long diffStashed  = currentStashedCount - lastStashedCount;
+            long diffDeferred = currentDeferredCount - lastDeferredCount;
 
             if (diffCount == 0 && diffSuccess == 0 && diffFailed == 0 && diffStashed == 0 && diffDeferred == 0) {
                 return;
             }
 
-            lastIntervalCount        = totalCount;
-            lastIntervalSuccessCount = totalSuccessCount;
-            lastIntervalFailedCount  = totalFailedCount;
-            lastStashedCount         = totalStashedCount;
-            lastDeferredCount        = totalDeferredCount;
+            lastIntervalCount        = currentTotalCount;
+            lastIntervalSuccessCount = currentSuccessCount;
+            lastIntervalFailedCount  = currentFailedCount;
+            lastStashedCount         = currentStashedCount;
+            lastDeferredCount        = currentDeferredCount;
 
             if (statusLogEnabled) {
                 String finalPath  = "";
@@ -475,6 +477,12 @@ public void logFailedEventJSON(Collection<String> events, Throwable excp) {
     }
 
     private void logAuditStatus(long diffTime, long diffCount, long diffSuccess, long diffFailed, long diffStashed, long diffDeferred, String finalPath) {
+        long currentTotalCount        = totalCount.get();
+        long currentTotalSuccessCount = totalSuccessCount.get();
+        long currentTotalFailedCount  = totalFailedCount.get();
+        long currentTotalStashedCount = totalStashedCount.get();
+        long currentTotalDeferredCount = totalDeferredCount.get();
+
         String msg = "Audit Status Log: name="
                 + getName()
                 + finalPath
@@ -489,14 +497,14 @@ private void logAuditStatus(long diffTime, long diffCount, long diffSuccess, lon
                 + (diffDeferred > 0 ? (", deferredCount=" + diffDeferred)
                 : "")
                 + ", totalEvents="
-                + totalCount
-                + (totalSuccessCount > 0 ? (", totalSuccessCount=" + totalSuccessCount)
+                + currentTotalCount
+                + (currentTotalSuccessCount > 0 ? (", totalSuccessCount=" + currentTotalSuccessCount)
                 : "")
-                + (totalFailedCount > 0 ? (", totalFailedCount=" + totalFailedCount)
+                + (currentTotalFailedCount > 0 ? (", totalFailedCount=" + currentTotalFailedCount)
                 : "")
-                + (totalStashedCount > 0 ? (", totalStashedCount=" + totalStashedCount)
+                + (currentTotalStashedCount > 0 ? (", totalStashedCount=" + currentTotalStashedCount)
                 : "")
-                + (totalDeferredCount > 0 ? (", totalDeferredCount=" + totalDeferredCount)
+                + (currentTotalDeferredCount > 0 ? (", totalDeferredCount=" + currentTotalDeferredCount)
                 : "");
         LOG.info(msg);
     }

agents-audit/core/src/main/java/org/apache/ranger/audit/provider/DummyAuditProvider.java

@@ -40,6 +40,11 @@ public boolean log(Collection<AuditEventBase> events) {
         return true;
     }
 
+    @Override
+    public boolean log(Collection<AuditEventBase> events, String batchKey) {
+        return log(events);
+    }
+
     @Override
     public boolean logJSON(String event) {
         AuditEventBase eventObj = MiscUtil.fromJson(event, AuthzAuditEvent.class);

agents-audit/core/src/main/java/org/apache/ranger/audit/utils/AbstractRangerAuditWriter.java

@@ -133,13 +133,15 @@ public void createFileSystemFolders() throws Exception {
 
         String defaultPath = fullPath;
 
+        fileSystemScheme = getFileSystemScheme();
+
         conf = createConfiguration();
 
         URI uri = URI.create(fullPath);
 
-        fileSystem       = FileSystem.get(uri, conf);
-        auditPath        = new Path(fullPath);
-        fileSystemScheme = getFileSystemScheme();
+        fileSystem = FileSystem.get(uri, conf);
+
+        auditPath  = new Path(fullPath);
 
         logger.info("Checking whether log file exists. {} Path={}, UGI={}", fileSystemScheme, fullPath, MiscUtil.getUGILoginUser());
 
@@ -195,6 +197,9 @@ public void createParents(Path pathLogfile, FileSystem fileSystem) throws Except
 
         if (parentPath != null && fileSystem != null && !fileSystem.exists(parentPath)) {
             fileSystem.mkdirs(parentPath);
+            logger.info("Successfully created parent folder: {}", parentPath);
+        } else {
+            logger.info("Parent folder already exists or not required: {}", parentPath);
         }
     }
 
@@ -308,14 +313,17 @@ public PrintWriter createWriter() throws Exception {
 
             if (!appendMode) {
                 // Create the file to write
-                logger.info("Creating new log file. auditPath = {}", fullPath);
-
                 createFileSystemFolders();
 
+                logger.info("Creating new log file. fullPath = {}", fullPath);
+
                 ostream = fileSystem.create(auditPath);
+                logger.info("Successfully created {} output stream for file: {}", fileSystemScheme, fullPath);
             }
             logWriter             = new PrintWriter(ostream);
             isHFlushCapableStream = ostream.hasCapability(StreamCapabilities.HFLUSH);
+
+            logger.info("{} audit writer initialized successfully. File: {}, HFlush capable: {}", fileSystemScheme, fullPath, isHFlushCapableStream);
         }
 
         logger.debug("<== AbstractRangerAuditWriter.createWriter()");

agents-audit/core/src/main/java/org/apache/ranger/audit/utils/RangerJSONAuditWriter.java

@@ -102,9 +102,13 @@ public synchronized boolean logJSON(final Collection<String> events) throws Exce
                 } else {
                     out1 = getLogFileStream();
 
+                    logger.debug("Writing {} audit events to HDFS file: {}", events.size(), currentFileName);
+
                     for (String event : events) {
                         out1.println(event);
                     }
+
+                    logger.debug("Successfully wrote {} audit events to HDFS", events.size());
                 }
 
                 return out1;