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/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

@@ -74,16 +74,16 @@ public abstract class BaseAuditHandler implements AuditHandler {
 
     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;
+    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);
+    AtomicLong lastIntervalCount       = new AtomicLong(0);
+    AtomicLong lastIntervalSuccessCount = new AtomicLong(0);
+    AtomicLong lastIntervalFailedCount = new AtomicLong(0);
+    AtomicLong lastStashedCount        = new AtomicLong(0);
+    AtomicLong lastDeferredCount       = 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();
@@ -237,61 +237,51 @@ 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;
+        return lastStashedCount.get();
     }
 
     public long getTotalDeferredCount() {
-        return totalDeferredCount;
+        return totalDeferredCount.get();
     }
 
     public long getLastDeferredCount() {
-        return lastDeferredCount;
+        return lastDeferredCount.get();
     }
 
     public boolean isStatusLogEnabled() {
@@ -312,21 +302,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.get();
+            long diffSuccess  = currentSuccessCount - lastIntervalSuccessCount.get();
+            long diffFailed   = currentFailedCount - lastIntervalFailedCount.get();
+            long diffStashed  = currentStashedCount - lastStashedCount.get();
+            long diffDeferred = currentDeferredCount - lastDeferredCount.get();
 
             if (diffCount == 0 && diffSuccess == 0 && diffFailed == 0 && diffStashed == 0 && diffDeferred == 0) {
                 return;
             }
 
-            lastIntervalCount        = totalCount;
-            lastIntervalSuccessCount = totalSuccessCount;
-            lastIntervalFailedCount  = totalFailedCount;
-            lastStashedCount         = totalStashedCount;
-            lastDeferredCount        = totalDeferredCount;
+            lastIntervalCount.set(currentTotalCount);
+            lastIntervalSuccessCount.set(currentSuccessCount);
+            lastIntervalFailedCount.set(currentFailedCount);
+            lastStashedCount.set(currentStashedCount);
+            lastDeferredCount.set(currentDeferredCount);
 
             if (statusLogEnabled) {
                 String finalPath  = "";
@@ -475,6 +471,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 +491,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/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;

agents-audit/dest-auditserver/pom.xml

@@ -0,0 +1,63 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.ranger</groupId>
+        <artifactId>ranger</artifactId>
+        <version>3.0.0-SNAPSHOT</version>
+        <relativePath>../..</relativePath>
+    </parent>
+    <artifactId>ranger-audit-dest-auditserver</artifactId>
+    <packaging>jar</packaging>
+    <name>Ranger Audit Destination - auditserver</name>
+    <description>Ranger Audit Destination - auditserver</description>
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <securesm.version>1.2</securesm.version>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.ranger</groupId>
+            <artifactId>ranger-audit-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>${slf4j.version}</version>
+        </dependency>
+
+        <!-- Test -->
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>log4j-over-slf4j</artifactId>
+            <version>${slf4j.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.testng</groupId>
+            <artifactId>testng</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>