You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ranger.apache.org by rm...@apache.org on 2021/04/06 06:56:31 UTC
[ranger] branch ranger-2.2 updated: RANGER-1837:Enhance Ranger
Audit to HDFS to support ORC file format
This is an automated email from the ASF dual-hosted git repository.
rmani pushed a commit to branch ranger-2.2
in repository https://gitbox.apache.org/repos/asf/ranger.git
The following commit(s) were added to refs/heads/ranger-2.2 by this push:
new 57b6f98 RANGER-1837:Enhance Ranger Audit to HDFS to support ORC file format
57b6f98 is described below
commit 57b6f98a1457ebef45dedfb9aa5bde9f09638c4b
Author: Ramesh Mani <rm...@cloudera.com>
AuthorDate: Mon Apr 5 23:56:15 2021 -0700
RANGER-1837:Enhance Ranger Audit to HDFS to support ORC file format
---
agents-audit/pom.xml | 11 +
.../audit/destination/HDFSAuditDestination.java | 324 +++------------
.../apache/ranger/audit/provider/AuditHandler.java | 4 +-
.../audit/provider/AuditProviderFactory.java | 29 +-
.../ranger/audit/provider/AuditWriterFactory.java | 117 ++++++
.../ranger/audit/provider/BaseAuditHandler.java | 7 +
.../ranger/audit/provider/DummyAuditProvider.java | 9 +
.../org/apache/ranger/audit/provider/MiscUtil.java | 13 +
.../audit/provider/MultiDestAuditProvider.java | 14 +
.../audit/queue/AuditFileCacheProviderSpool.java | 9 +-
.../apache/ranger/audit/queue/AuditFileQueue.java | 126 ++++++
...ProviderSpool.java => AuditFileQueueSpool.java} | 316 +++++++++-----
.../audit/utils/AbstractRangerAuditWriter.java | 353 ++++++++++++++++
.../org/apache/ranger/audit/utils/ORCFileUtil.java | 454 +++++++++++++++++++++
.../ranger/audit/utils/RangerAuditWriter.java | 39 ++
.../ranger/audit/utils/RangerJSONAuditWriter.java | 175 ++++++++
.../ranger/audit/utils/RangerORCAuditWriter.java | 189 +++++++++
17 files changed, 1809 insertions(+), 380 deletions(-)
diff --git a/agents-audit/pom.xml b/agents-audit/pom.xml
index bc43048..d67b925 100644
--- a/agents-audit/pom.xml
+++ b/agents-audit/pom.xml
@@ -264,5 +264,16 @@
<artifactId>log4j-api</artifactId>
<version>${log4j.core.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hive</groupId>
+ <artifactId>hive-exec</artifactId>
+ <version>${hive.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>*</groupId>
+ <artifactId>*</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
</dependencies>
</project>
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/destination/HDFSAuditDestination.java b/agents-audit/src/main/java/org/apache/ranger/audit/destination/HDFSAuditDestination.java
index 5e6f402..ec9dc4b 100644
--- a/agents-audit/src/main/java/org/apache/ranger/audit/destination/HDFSAuditDestination.java
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/destination/HDFSAuditDestination.java
@@ -19,25 +19,20 @@
package org.apache.ranger.audit.destination;
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.io.PrintWriter;
-import java.net.URI;
+import java.io.File;
import java.security.PrivilegedAction;
-import java.security.PrivilegedExceptionAction;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
-import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.StreamCapabilities;
import org.apache.ranger.audit.model.AuditEventBase;
+import org.apache.ranger.audit.provider.AuditWriterFactory;
import org.apache.ranger.audit.provider.MiscUtil;
-import org.apache.ranger.audit.utils.RollingTimeUtil;
+import org.apache.ranger.audit.utils.RangerAuditWriter;
/**
* This class write the logs to local file
@@ -46,90 +41,24 @@ public class HDFSAuditDestination extends AuditDestination {
private static final Log logger = LogFactory
.getLog(HDFSAuditDestination.class);
- public static final String PROP_HDFS_DIR = "dir";
- public static final String PROP_HDFS_SUBDIR = "subdir";
- public static final String PROP_HDFS_FILE_NAME_FORMAT = "filename.format";
- public static final String PROP_HDFS_ROLLOVER = "file.rollover.sec";
- public static final String PROP_HDFS_ROLLOVER_PERIOD = "file.rollover.period";
-
- int fileRolloverSec = 24 * 60 * 60; // In seconds
-
- private String logFileNameFormat;
-
- private String rolloverPeriod;
-
- boolean initDone = false;
-
- private String logFolder;
-
- private PrintWriter logWriter = null;
- volatile FSDataOutputStream ostream = null; // output stream wrapped in logWriter
-
- private String currentFileName;
-
- private boolean isStopped = false;
-
- private RollingTimeUtil rollingTimeUtil = null;
-
- private Date nextRollOverTime = null;
-
- private boolean rollOverByDuration = false;
-
- private boolean isHFlushCapableStream = false;
+ private Map<String, String> auditConfigs = null;
+ private String auditProviderName = null;
+ private RangerAuditWriter auditWriter = null;
+ private boolean initDone = false;
+ private boolean isStopped = false;
@Override
public void init(Properties prop, String propPrefix) {
super.init(prop, propPrefix);
-
- // Initialize properties for this class
- // Initial folder and file properties
- String logFolderProp = MiscUtil.getStringProperty(props, propPrefix
- + "." + PROP_HDFS_DIR);
- if (logFolderProp == null || logFolderProp.isEmpty()) {
- logger.fatal("File destination folder is not configured. Please set "
- + propPrefix + "." + PROP_HDFS_DIR + ". name=" + getName());
- return;
- }
-
- String logSubFolder = MiscUtil.getStringProperty(props, propPrefix
- + "." + PROP_HDFS_SUBDIR);
- if (logSubFolder == null || logSubFolder.isEmpty()) {
- logSubFolder = "%app-type%/%time:yyyyMMdd%";
- }
-
- logFileNameFormat = MiscUtil.getStringProperty(props, propPrefix + "."
- + PROP_HDFS_FILE_NAME_FORMAT);
- fileRolloverSec = MiscUtil.getIntProperty(props, propPrefix + "."
- + PROP_HDFS_ROLLOVER, fileRolloverSec);
-
- if (logFileNameFormat == null || logFileNameFormat.isEmpty()) {
- logFileNameFormat = "%app-type%_ranger_audit_%hostname%" + ".log";
- }
-
- logFolder = logFolderProp + "/" + logSubFolder;
- logger.info("logFolder=" + logFolder + ", destName=" + getName());
- logger.info("logFileNameFormat=" + logFileNameFormat + ", destName="
- + getName());
- logger.info("config=" + configProps.toString());
-
- rolloverPeriod = MiscUtil.getStringProperty(props, propPrefix + "." + PROP_HDFS_ROLLOVER_PERIOD);
- rollingTimeUtil = RollingTimeUtil.getInstance();
-
- //file.rollover.period is used for rolling over. If it could compute the next roll over time using file.rollover.period
- //it fall back to use file.rollover.sec for find next rollover time. If still couldn't find default will be 1day window
- //for rollover.
- if(StringUtils.isEmpty(rolloverPeriod) ) {
- rolloverPeriod = rollingTimeUtil.convertRolloverSecondsToRolloverPeriod(fileRolloverSec);
- }
+ this.auditProviderName = getName();
+ this.auditConfigs = configProps;
try {
- nextRollOverTime = rollingTimeUtil.computeNextRollingTime(rolloverPeriod);
- } catch ( Exception e) {
- logger.warn("Rollover by file.rollover.period failed...will be using the file.rollover.sec for hdfs audit file rollover...",e);
- rollOverByDuration = true;
- nextRollOverTime = rollOverByDuration();
+ this.auditWriter = getWriter();
+ this.initDone = true;
+ } catch (Exception e) {
+ logger.error("Error while getting Audit writer", e);
}
- initDone = true;
}
@Override
@@ -146,33 +75,10 @@ public class HDFSAuditDestination extends AuditDestination {
logError("log() called after stop was requested. name=" + getName());
return false;
}
-
- PrintWriter out = null;
try {
- if (logger.isDebugEnabled()) {
- logger.debug("UGI=" + MiscUtil.getUGILoginUser()
- + ". Will write to HDFS file=" + currentFileName);
- }
-
- out = MiscUtil.executePrivilegedAction(new PrivilegedExceptionAction<PrintWriter>() {
- @Override
- public PrintWriter run() throws Exception {
- PrintWriter out = getLogFileStream();
- for (String event : events) {
- out.println(event);
- }
- return out;
- };
- });
-
- // flush and check the stream for errors
- if (out.checkError()) {
- // In theory, this count may NOT be accurate as part of the messages may have been successfully written.
- // However, in practice, since client does buffering, either all of none would succeed.
+ boolean ret = auditWriter.log(events);
+ if (!ret) {
addDeferredCount(events.size());
- out.close();
- logWriter = null;
- ostream = null;
return false;
}
} catch (Throwable t) {
@@ -181,7 +87,7 @@ public class HDFSAuditDestination extends AuditDestination {
return false;
} finally {
logger.info("Flushing HDFS audit. Event Size:" + events.size());
- if (out != null) {
+ if (auditWriter != null) {
flush();
}
}
@@ -190,12 +96,40 @@ public class HDFSAuditDestination extends AuditDestination {
}
@Override
+ synchronized public boolean logFile(final File file) {
+ logStatusIfRequired();
+ if (!initDone) {
+ return false;
+ }
+ if (isStopped) {
+ logError("log() called after stop was requested. name=" + getName());
+ return false;
+ }
+
+ try {
+ boolean ret = auditWriter.logFile(file);
+ if (!ret) {
+ return false;
+ }
+ } catch (Throwable t) {
+ logError("Error writing to log file.", t);
+ return false;
+ } finally {
+ logger.info("Flushing HDFS audit. File:" + file.getAbsolutePath() + file.getName());
+ if (auditWriter != null) {
+ flush();
+ }
+ }
+ return true;
+ }
+
+ @Override
public void flush() {
logger.info("Flush called. name=" + getName());
MiscUtil.executePrivilegedAction(new PrivilegedAction<Void>() {
@Override
public Void run() {
- hflush();
+ auditWriter.flush();
return null;
}
});
@@ -244,158 +178,14 @@ public class HDFSAuditDestination extends AuditDestination {
@Override
synchronized public void stop() {
- isStopped = true;
- if (logWriter != null) {
- try {
- logWriter.flush();
- logWriter.close();
- } catch (Throwable t) {
- logger.error("Error on closing log writter. Exception will be ignored. name="
- + getName() + ", fileName=" + currentFileName);
- }
- logWriter = null;
- ostream = null;
- }
+ auditWriter.stop();
logStatus();
+ isStopped = true;
}
- // Helper methods in this class
- synchronized private PrintWriter getLogFileStream() throws Exception {
- closeFileIfNeeded();
-
- // Either there are no open log file or the previous one has been rolled
- // over
- if (logWriter == null) {
- Date currentTime = new Date();
- // Create a new file
- String fileName = MiscUtil.replaceTokens(logFileNameFormat,
- currentTime.getTime());
- String parentFolder = MiscUtil.replaceTokens(logFolder,
- currentTime.getTime());
- Configuration conf = createConfiguration();
-
- String fullPath = parentFolder + Path.SEPARATOR + fileName;
- String defaultPath = fullPath;
- URI uri = URI.create(fullPath);
- FileSystem fileSystem = FileSystem.get(uri, conf);
-
- Path hdfPath = new Path(fullPath);
- logger.info("Checking whether log file exists. hdfPath=" + fullPath + ", UGI=" + MiscUtil.getUGILoginUser());
- int i = 0;
- while (fileSystem.exists(hdfPath)) {
- i++;
- int lastDot = defaultPath.lastIndexOf('.');
- String baseName = defaultPath.substring(0, lastDot);
- String extension = defaultPath.substring(lastDot);
- fullPath = baseName + "." + i + extension;
- hdfPath = new Path(fullPath);
- logger.info("Checking whether log file exists. hdfPath="
- + fullPath);
- }
- logger.info("Log file doesn't exists. Will create and use it. hdfPath="
- + fullPath);
- // Create parent folders
- createParents(hdfPath, fileSystem);
-
- // Create the file to write
- logger.info("Creating new log file. hdfPath=" + fullPath);
- ostream = fileSystem.create(hdfPath);
- logWriter = new PrintWriter(ostream);
- currentFileName = fullPath;
- isHFlushCapableStream = ostream.hasCapability(StreamCapabilities.HFLUSH);
- }
- return logWriter;
- }
-
- Configuration createConfiguration() {
- Configuration conf = new Configuration();
- for (Map.Entry<String, String> entry : configProps.entrySet()) {
- String key = entry.getKey();
- String value = entry.getValue();
- // for ease of install config file may contain properties with empty value, skip those
- if (StringUtils.isNotEmpty(value)) {
- conf.set(key, value);
- }
- logger.info("Adding property to HDFS config: " + key + " => " + value);
- }
-
- logger.info("Returning HDFS Filesystem Config: " + conf.toString());
- return conf;
- }
-
- private void createParents(Path pathLogfile, FileSystem fileSystem)
- throws Exception {
- logger.info("Creating parent folder for " + pathLogfile);
- Path parentPath = pathLogfile != null ? pathLogfile.getParent() : null;
-
- if (parentPath != null && fileSystem != null
- && !fileSystem.exists(parentPath)) {
- fileSystem.mkdirs(parentPath);
- }
- }
-
- private void closeFileIfNeeded() throws FileNotFoundException, IOException {
- if (logWriter == null) {
- return;
- }
-
- if ( System.currentTimeMillis() > nextRollOverTime.getTime() ) {
- logger.info("Closing file. Rolling over. name=" + getName()
- + ", fileName=" + currentFileName);
- try {
- logWriter.flush();
- logWriter.close();
- } catch (Throwable t) {
- logger.error("Error on closing log writter. Exception will be ignored. name="
- + getName() + ", fileName=" + currentFileName);
- }
-
- logWriter = null;
- ostream = null;
- currentFileName = null;
-
- if (!rollOverByDuration) {
- try {
- if(StringUtils.isEmpty(rolloverPeriod) ) {
- rolloverPeriod = rollingTimeUtil.convertRolloverSecondsToRolloverPeriod(fileRolloverSec);
- }
- nextRollOverTime = rollingTimeUtil.computeNextRollingTime(rolloverPeriod);
- } catch ( Exception e) {
- logger.warn("Rollover by file.rollover.period failed...will be using the file.rollover.sec for hdfs audit file rollover...",e);
- nextRollOverTime = rollOverByDuration();
- }
- } else {
- nextRollOverTime = rollOverByDuration();
- }
- }
- }
-
- private void hflush() {
- if (ostream != null) {
- try {
- synchronized (this) {
- if (ostream != null)
- // 1) PrinterWriter does not have bufferring of its own so
- // we need to flush its underlying stream
- // 2) HDFS flush() does not really flush all the way to disk.
- if (isHFlushCapableStream) {
- //Checking HFLUSH capability of the stream because of HADOOP-13327.
- //For S3 filesystem, hflush throws UnsupportedOperationException and hence we call flush.
- ostream.hflush();
- } else {
- ostream.flush();
- }
- logger.info("Flush HDFS audit logs completed.....");
- }
- } catch (IOException e) {
- logger.error("Error on flushing log writer: " + e.getMessage() +
- "\nException will be ignored. name=" + getName() + ", fileName=" + currentFileName);
- }
- }
- }
-
- private Date rollOverByDuration() {
- long rollOverTime = rollingTimeUtil.computeNextRollingTime(fileRolloverSec,nextRollOverTime);
- return new Date(rollOverTime);
+ public RangerAuditWriter getWriter() throws Exception {
+ AuditWriterFactory auditWriterFactory = AuditWriterFactory.getInstance();
+ auditWriterFactory.init(props, propPrefix, auditProviderName, auditConfigs);
+ return auditWriterFactory.getAuditWriter();
}
}
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/provider/AuditHandler.java b/agents-audit/src/main/java/org/apache/ranger/audit/provider/AuditHandler.java
index 4ce31dd..dd02255 100644
--- a/agents-audit/src/main/java/org/apache/ranger/audit/provider/AuditHandler.java
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/provider/AuditHandler.java
@@ -18,6 +18,7 @@
package org.apache.ranger.audit.provider;
+import java.io.File;
import java.util.Collection;
import java.util.Properties;
@@ -28,7 +29,8 @@ public interface AuditHandler {
boolean log(Collection<AuditEventBase> events);
boolean logJSON(String event);
- boolean logJSON(Collection<String> events);
+ boolean logJSON(Collection<String> events);
+ boolean logFile(File file);
void init(Properties prop);
void init(Properties prop, String basePropertyName);
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/provider/AuditProviderFactory.java b/agents-audit/src/main/java/org/apache/ranger/audit/provider/AuditProviderFactory.java
index 6b7f4b0..7a3c7f6 100644
--- a/agents-audit/src/main/java/org/apache/ranger/audit/provider/AuditProviderFactory.java
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/provider/AuditProviderFactory.java
@@ -35,6 +35,7 @@ import org.apache.ranger.audit.provider.kafka.KafkaAuditProvider;
import org.apache.ranger.audit.provider.solr.SolrAuditProvider;
import org.apache.ranger.audit.queue.AuditAsyncQueue;
import org.apache.ranger.audit.queue.AuditBatchQueue;
+import org.apache.ranger.audit.queue.AuditFileQueue;
import org.apache.ranger.audit.queue.AuditQueue;
import org.apache.ranger.audit.queue.AuditSummaryQueue;
@@ -58,6 +59,8 @@ public class AuditProviderFactory {
public static final String AUDIT_DEST_BASE = "xasecure.audit.destination";
public static final String AUDIT_SHUTDOWN_HOOK_MAX_WAIT_SEC = "xasecure.audit.shutdown.hook.max.wait.seconds";
public static final String AUDIT_IS_FILE_CACHE_PROVIDER_ENABLE_PROP = "xasecure.audit.provider.filecache.is.enabled";
+ public static final String FILE_QUEUE_TYPE = "filequeue";
+ public static final String DEFAULT_QUEUE_TYPE = "memoryqueue";
public static final int AUDIT_SHUTDOWN_HOOK_MAX_WAIT_SEC_DEFAULT = 30;
public static final int AUDIT_ASYNC_MAX_QUEUE_SIZE_DEFAULT = 10 * 1024;
@@ -423,7 +426,7 @@ public class AuditProviderFactory {
} else if (providerName.equalsIgnoreCase("log4j")) {
provider = new Log4JAuditDestination();
} else if (providerName.equalsIgnoreCase("batch")) {
- provider = new AuditBatchQueue(consumer);
+ provider = getAuditProvider(props, propPrefix, consumer);
} else if (providerName.equalsIgnoreCase("async")) {
provider = new AuditAsyncQueue(consumer);
} else {
@@ -441,6 +444,30 @@ public class AuditProviderFactory {
return provider;
}
+ private AuditHandler getAuditProvider(Properties props, String propPrefix, AuditHandler consumer) {
+ AuditHandler ret = null;
+ String queueType = MiscUtil.getStringProperty(props, propPrefix + "." + "queuetype", DEFAULT_QUEUE_TYPE);
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("==> AuditProviderFactory.getAuditProvider() propPerfix= " + propPrefix + ", " + " queueType= " + queueType);
+ }
+
+ if (FILE_QUEUE_TYPE.equalsIgnoreCase(queueType)) {
+ AuditFileQueue auditFileQueue = new AuditFileQueue(consumer);
+ String propPrefixFileQueue = propPrefix + "." + FILE_QUEUE_TYPE;
+ auditFileQueue.init(props, propPrefixFileQueue);
+ ret = new AuditBatchQueue(auditFileQueue);
+ } else {
+ ret = new AuditBatchQueue(consumer);
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("<== AuditProviderFactory.getAuditProvider()");
+ }
+
+ return ret;
+ }
+
private AuditHandler getDefaultProvider() {
return new DummyAuditProvider();
}
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/provider/AuditWriterFactory.java b/agents-audit/src/main/java/org/apache/ranger/audit/provider/AuditWriterFactory.java
new file mode 100644
index 0000000..6ef2255
--- /dev/null
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/provider/AuditWriterFactory.java
@@ -0,0 +1,117 @@
+package org.apache.ranger.audit.provider;
+
+/*
+ * 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.
+ */
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.ranger.audit.utils.RangerAuditWriter;
+
+import java.util.Map;
+import java.util.Properties;
+
+public class AuditWriterFactory {
+ private static final Log logger = LogFactory.getLog(AuditWriterFactory.class);
+ public static final String AUDIT_FILETYPE_DEFAULT = "json";
+ public static final String AUDIT_JSON_FILEWRITER_IMPL = "org.apache.ranger.audit.utils.RangerJSONAuditWriter";
+ public static final String AUDIT_ORC_FILEWRITER_IMPL = "org.apache.ranger.audit.utils.RangerORCAuditWriter";
+
+ public Map<String,String> auditConfigs = null;
+ public Properties props = null;
+ public String propPrefix = null;
+ public String auditProviderName = null;
+ public RangerAuditWriter auditWriter = null;
+ private static volatile AuditWriterFactory me = null;
+
+ public static AuditWriterFactory getInstance() {
+ AuditWriterFactory auditWriter = me;
+ if (auditWriter == null) {
+ synchronized (AuditWriterFactory.class) {
+ auditWriter = me;
+ if (auditWriter == null) {
+ me = auditWriter = new AuditWriterFactory();
+ }
+ }
+ }
+ return auditWriter;
+ }
+
+ public void init(Properties props, String propPrefix, String auditProviderName, Map<String,String> auditConfigs) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> AuditWriterFactory.init()");
+ }
+ this.props = props;
+ this.propPrefix = propPrefix;
+ this.auditProviderName = auditProviderName;
+ this.auditConfigs = auditConfigs;
+ String auditFileType = MiscUtil.getStringProperty(props, propPrefix + ".filetype", AUDIT_FILETYPE_DEFAULT);
+ String writerClass = MiscUtil.getStringProperty(props, propPrefix + ".filewriter.impl");
+
+ auditWriter = StringUtils.isEmpty(writerClass) ? createWriter(getDefaultWriter(auditFileType)) : createWriter(writerClass);
+
+ if (auditWriter != null) {
+ auditWriter.init(props, propPrefix, auditProviderName, auditConfigs);
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== AuditWriterFactory.init() :" + auditWriter.getClass().getName());
+ }
+ }
+
+ public RangerAuditWriter createWriter(String writerClass) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> AuditWriterFactory.createWriter()");
+ }
+ RangerAuditWriter ret = null;
+ try {
+ Class<RangerAuditWriter> cls = (Class<RangerAuditWriter>) Class.forName(writerClass);
+ ret = cls.newInstance();
+ } catch (Exception e) {
+ throw e;
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== AuditWriterFactory.createWriter()");
+ }
+ return ret;
+ }
+
+ public String getDefaultWriter(String auditFileType) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> AuditWriterFactory.getDefaultWriter()");
+ }
+ String ret = null;
+ switch (auditFileType) {
+ case "orc":
+ ret = AUDIT_ORC_FILEWRITER_IMPL;
+ break;
+ case "json":
+ ret = AUDIT_JSON_FILEWRITER_IMPL;
+ break;
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== AuditWriterFactory.getDefaultWriter() :" + ret);
+ }
+ return ret;
+ }
+
+ public RangerAuditWriter getAuditWriter(){
+ return this.auditWriter;
+ }
+}
\ No newline at end of file
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/provider/BaseAuditHandler.java b/agents-audit/src/main/java/org/apache/ranger/audit/provider/BaseAuditHandler.java
index 54f3764..1987856 100644
--- a/agents-audit/src/main/java/org/apache/ranger/audit/provider/BaseAuditHandler.java
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/provider/BaseAuditHandler.java
@@ -25,6 +25,8 @@ import org.apache.ranger.audit.model.AuthzAuditEvent;
import com.google.gson.GsonBuilder;
+//import java.io.File;
+import java.io.File;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
@@ -191,6 +193,11 @@ public abstract class BaseAuditHandler implements AuditHandler {
return log(eventList);
}
+ @Override
+ public boolean logFile(File file) {
+ return logFile(file);
+ }
+
public String getParentPath() {
return parentPath;
}
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/provider/DummyAuditProvider.java b/agents-audit/src/main/java/org/apache/ranger/audit/provider/DummyAuditProvider.java
index 05f882f..cbd25ab 100644
--- a/agents-audit/src/main/java/org/apache/ranger/audit/provider/DummyAuditProvider.java
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/provider/DummyAuditProvider.java
@@ -17,6 +17,7 @@
*/
package org.apache.ranger.audit.provider;
+import java.io.File;
import java.util.Collection;
import java.util.Properties;
@@ -103,4 +104,12 @@ public class DummyAuditProvider implements AuditHandler {
return this.getClass().getName();
}
+ /* (non-Javadoc)
+ * @see org.apache.ranger.audit.provider.AuditProvider#getAuditFileType()
+ */
+ @Override
+ public boolean logFile(File file) {
+ return logFile(file);
+ }
+
}
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/provider/MiscUtil.java b/agents-audit/src/main/java/org/apache/ranger/audit/provider/MiscUtil.java
index e2b7448..f58b813 100644
--- a/agents-audit/src/main/java/org/apache/ranger/audit/provider/MiscUtil.java
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/provider/MiscUtil.java
@@ -339,6 +339,19 @@ public class MiscUtil {
return ret;
}
+ public static String getStringProperty(Properties props, String propName, String defValue) {
+ String ret = defValue;
+
+ if (props != null && propName != null) {
+ String val = props.getProperty(propName);
+ if (val != null) {
+ ret = val;
+ }
+ }
+
+ return ret;
+ }
+
public static boolean getBooleanProperty(Properties props, String propName,
boolean defValue) {
boolean ret = defValue;
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/provider/MultiDestAuditProvider.java b/agents-audit/src/main/java/org/apache/ranger/audit/provider/MultiDestAuditProvider.java
index 282f5ab..b829ac5 100644
--- a/agents-audit/src/main/java/org/apache/ranger/audit/provider/MultiDestAuditProvider.java
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/provider/MultiDestAuditProvider.java
@@ -17,6 +17,7 @@
*/
package org.apache.ranger.audit.provider;
+import java.io.File;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
@@ -155,6 +156,19 @@ public class MultiDestAuditProvider extends BaseAuditHandler {
return true;
}
+
+ @Override
+ public boolean logFile(File file) {
+ for (AuditHandler provider : mProviders) {
+ try {
+ provider.logFile(file);
+ } catch (Throwable excp) {
+ logFailedEventJSON(file.getAbsolutePath(), excp);
+ }
+ }
+ return true;
+ }
+
@Override
public void start() {
for (AuditHandler provider : mProviders) {
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileCacheProviderSpool.java b/agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileCacheProviderSpool.java
index 41513ba..f6951c6 100644
--- a/agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileCacheProviderSpool.java
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileCacheProviderSpool.java
@@ -56,10 +56,10 @@ public class AuditFileCacheProviderSpool implements Runnable {
public static final String PROP_FILE_SPOOL_FILE_ROLLOVER = "filespool.file.rollover.sec";
public static final String PROP_FILE_SPOOL_INDEX_FILE = "filespool.index.filename";
public static final String PROP_FILE_SPOOL_DEST_RETRY_MS = "filespool.destination.retry.ms";
+ public static final String PROP_FILE_SPOOL_BATCH_SIZE = "filespool.buffer.size";
public static final String AUDIT_IS_FILE_CACHE_PROVIDER_ENABLE_PROP = "xasecure.audit.provider.filecache.is.enabled";
public static final String FILE_CACHE_PROVIDER_NAME = "AuditFileCacheProviderSpool";
- public static final int AUDIT_BATCH_SIZE_DEFAULT = 1000;
AuditHandler consumerProvider = null;
@@ -79,6 +79,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
int fileRolloverSec = 24 * 60 * 60; // In seconds
int maxArchiveFiles = 100;
int errorLogIntervalMS = 30 * 1000; // Every 30 seconds
+ int auditBatchSize = 1000;
long lastErrorLogMS = 0;
boolean isAuditFileCacheProviderEnabled = false;
boolean closeFile = false;
@@ -275,6 +276,10 @@ public class AuditFileCacheProviderSpool implements Runnable {
+ FILE_CACHE_PROVIDER_NAME, t);
return false;
}
+
+ auditBatchSize = MiscUtil.getIntProperty(props, propPrefix
+ + "." + PROP_FILE_SPOOL_BATCH_SIZE, auditBatchSize);
+
initDone = true;
logger.debug("<== AuditFileCacheProviderSpool.init()");
@@ -824,7 +829,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
AuditEventBase event = MiscUtil.fromJson(line, AuthzAuditEvent.class);
events.add(event);
- if (events.size() == AUDIT_BATCH_SIZE_DEFAULT) {
+ if (events.size() == auditBatchSize) {
boolean ret = sendEvent(events,
currentConsumerIndexRecord, currLine);
if (!ret) {
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileQueue.java b/agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileQueue.java
new file mode 100644
index 0000000..0b245c0
--- /dev/null
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileQueue.java
@@ -0,0 +1,126 @@
+/*
+ * 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.ranger.audit.queue;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.ranger.audit.model.AuditEventBase;
+import org.apache.ranger.audit.provider.AuditHandler;
+import org.apache.ranger.audit.provider.BaseAuditHandler;
+
+import java.util.Collection;
+import java.util.Properties;
+
+/*
+ AuditFileQueue class does the work of stashing the audit logs into Local Filesystem before sending it to the AuditBatchQueue Consumer
+*/
+
+public class AuditFileQueue extends BaseAuditHandler {
+ private static final Log logger = LogFactory.getLog(AuditFileQueue.class);
+
+ AuditFileQueueSpool fileSpooler = null;
+ AuditHandler consumer = null;
+
+ static final String DEFAULT_NAME = "batch";
+
+ public AuditFileQueue(AuditHandler consumer) {
+ this.consumer = consumer;
+ }
+
+ public void init(Properties prop, String basePropertyName) {
+ String propPrefix = "xasecure.audit.batch";
+ if (basePropertyName != null) {
+ propPrefix = basePropertyName;
+ }
+ super.init(prop, propPrefix);
+
+ //init AuditFileQueueSpooler thread to send Local logs to destination
+ fileSpooler = new AuditFileQueueSpool(consumer);
+ fileSpooler.init(prop,propPrefix);
+ }
+
+ @Override
+ public boolean log(AuditEventBase event) {
+ boolean ret = false;
+ if ( event != null) {
+ fileSpooler.stashLogs(event);
+ if (fileSpooler.isSpoolingSuccessful()) {
+ ret = true;
+ }
+ }
+ return ret;
+ }
+
+ @Override
+ public boolean log(Collection<AuditEventBase> events) {
+ boolean ret = true;
+ if ( events != null) {
+ for (AuditEventBase event : events) {
+ ret = log(event);
+ }
+ }
+ return ret;
+ }
+
+
+ @Override
+ public void start() {
+ // Start the consumer thread
+ if (consumer != null) {
+ consumer.start();
+ }
+ if (fileSpooler != null) {
+ // start AuditFileSpool thread
+ fileSpooler.start();
+ }
+ }
+
+ @Override
+ public void stop() {
+ logger.info("Stop called. name=" + getName());
+ if (consumer != null) {
+ consumer.stop();
+ }
+ }
+
+ @Override
+ public void waitToComplete() {
+ logger.info("waitToComplete called. name=" + getName());
+ if ( consumer != null) {
+ consumer.waitToComplete();
+ }
+ }
+
+ @Override
+ public void waitToComplete(long timeout) {
+ logger.info("waitToComplete called. name=" + getName());
+ if ( consumer != null) {
+ consumer.waitToComplete(timeout);
+ }
+ }
+
+ @Override
+ public void flush() {
+ logger.info("waitToComplete. name=" + getName());
+ if ( consumer != null) {
+ consumer.flush();
+ }
+ }
+
+}
\ No newline at end of file
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileCacheProviderSpool.java b/agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileQueueSpool.java
similarity index 76%
copy from agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileCacheProviderSpool.java
copy to agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileQueueSpool.java
index 41513ba..3284783 100644
--- a/agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileCacheProviderSpool.java
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/queue/AuditFileQueueSpool.java
@@ -1,21 +1,21 @@
/*
- * 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.
- */
+* 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.ranger.audit.queue;
@@ -25,12 +25,28 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.log4j.MDC;
import org.apache.ranger.audit.model.AuditEventBase;
-import org.apache.ranger.audit.provider.AuditHandler;
import org.apache.ranger.audit.model.AuthzAuditEvent;
+import org.apache.ranger.audit.provider.AuditHandler;
import org.apache.ranger.audit.provider.MiscUtil;
-import java.io.*;
-import java.util.*;
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileFilter;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.io.PrintWriter;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Properties;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
@@ -38,33 +54,31 @@ import java.util.concurrent.TimeUnit;
/**
* This class temporarily stores logs in Local file system before it despatches each logs in file to the AuditBatchQueue Consumer.
* This gets instantiated only when AuditFileCacheProvider is enabled (xasecure.audit.provider.filecache.is.enabled).
- * When AuditFileCacheProvider is all the logs are stored in local file system before sent to destination.
+ * When AuditFileCacheProvider is enabled all the logs are stored in local file system before sent to destination.
*/
-public class AuditFileCacheProviderSpool implements Runnable {
- private static final Log logger = LogFactory.getLog(AuditFileCacheProviderSpool.class);
+public class AuditFileQueueSpool implements Runnable {
+ private static final Log logger = LogFactory.getLog(AuditFileQueueSpool.class);
public enum SPOOL_FILE_STATUS {
pending, write_inprogress, read_inprogress, done
}
- public static final String PROP_FILE_SPOOL_LOCAL_DIR = "filespool.dir";
- public static final String PROP_FILE_SPOOL_LOCAL_FILE_NAME = "filespool.filename.format";
- public static final String PROP_FILE_SPOOL_ARCHIVE_DIR = "filespool.archive.dir";
- public static final String PROP_FILE_SPOOL_ARCHIVE_MAX_FILES_COUNT = "filespool.archive.max.files";
- public static final String PROP_FILE_SPOOL_FILENAME_PREFIX = "filespool.file.prefix";
- public static final String PROP_FILE_SPOOL_FILE_ROLLOVER = "filespool.file.rollover.sec";
- public static final String PROP_FILE_SPOOL_INDEX_FILE = "filespool.index.filename";
- public static final String PROP_FILE_SPOOL_DEST_RETRY_MS = "filespool.destination.retry.ms";
-
- public static final String AUDIT_IS_FILE_CACHE_PROVIDER_ENABLE_PROP = "xasecure.audit.provider.filecache.is.enabled";
- public static final String FILE_CACHE_PROVIDER_NAME = "AuditFileCacheProviderSpool";
- public static final int AUDIT_BATCH_SIZE_DEFAULT = 1000;
-
- AuditHandler consumerProvider = null;
-
- BlockingQueue<AuditIndexRecord> indexQueue = new LinkedBlockingQueue<AuditIndexRecord>();
- List<AuditIndexRecord> indexRecords = new ArrayList<AuditIndexRecord>();
+ public static final String PROP_FILE_SPOOL_LOCAL_DIR = "filespool.dir";
+ public static final String PROP_FILE_SPOOL_LOCAL_FILE_NAME = "filespool.filename.format";
+ public static final String PROP_FILE_SPOOL_ARCHIVE_DIR = "filespool.archive.dir";
+ public static final String PROP_FILE_SPOOL_ARCHIVE_MAX_FILES_COUNT = "filespool.archive.max.files";
+ public static final String PROP_FILE_SPOOL_FILENAME_PREFIX = "filespool.file.prefix";
+ public static final String PROP_FILE_SPOOL_FILE_ROLLOVER = "filespool.file.rollover.sec";
+ public static final String PROP_FILE_SPOOL_INDEX_FILE = "filespool.index.filename";
+ public static final String PROP_FILE_SPOOL_DEST_RETRY_MS = "filespool.destination.retry.ms";
+ public static final String PROP_FILE_SPOOL_BATCH_SIZE = "filespool.buffer.size";
+ public static final String FILE_QUEUE_PROVIDER_NAME = "AuditFileQueueSpool";
+ public static final String DEFAULT_AUDIT_FILE_TYPE = "json";
+
+ AuditHandler consumerProvider = null;
+ BlockingQueue<AuditIndexRecord> indexQueue = new LinkedBlockingQueue<AuditIndexRecord>();
+ List<AuditIndexRecord> indexRecords = new ArrayList<AuditIndexRecord>();
// Folder and File attributes
File logFolder = null;
@@ -74,6 +88,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
String indexFileName = null;
File indexFile = null;
String indexDoneFileName = null;
+ String auditFileType = null;
File indexDoneFile = null;
int retryDestinationMS = 30 * 1000; // Default 30 seconds
int fileRolloverSec = 24 * 60 * 60; // In seconds
@@ -84,6 +99,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
boolean closeFile = false;
boolean isPending = false;
long lastAttemptTime = 0;
+ long bufferSize = 1000;
boolean initDone = false;
PrintWriter logWriter = null;
@@ -100,7 +116,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
private Gson gson = null;
- public AuditFileCacheProviderSpool(AuditHandler consumerProvider) {
+ public AuditFileQueueSpool(AuditHandler consumerProvider) {
this.consumerProvider = consumerProvider;
}
@@ -109,7 +125,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
}
public boolean init(Properties props, String basePropertyName) {
- logger.debug("==> AuditFileCacheProviderSpool.init()");
+ logger.debug("==> AuditFileQueueSpool.init()");
if (initDone) {
logger.error("init() called more than once. queueProvider="
@@ -142,20 +158,19 @@ public class AuditFileCacheProviderSpool implements Runnable {
+ PROP_FILE_SPOOL_FILE_ROLLOVER, fileRolloverSec);
maxArchiveFiles = MiscUtil.getIntProperty(props, propPrefix + "."
+ PROP_FILE_SPOOL_ARCHIVE_MAX_FILES_COUNT, maxArchiveFiles);
- isAuditFileCacheProviderEnabled = MiscUtil.getBooleanProperty(props, AUDIT_IS_FILE_CACHE_PROVIDER_ENABLE_PROP, false);
logger.info("retryDestinationMS=" + retryDestinationMS
- + ", queueName=" + FILE_CACHE_PROVIDER_NAME);
+ + ", queueName=" + FILE_QUEUE_PROVIDER_NAME);
logger.info("fileRolloverSec=" + fileRolloverSec + ", queueName="
- + FILE_CACHE_PROVIDER_NAME);
+ + FILE_QUEUE_PROVIDER_NAME);
logger.info("maxArchiveFiles=" + maxArchiveFiles + ", queueName="
- + FILE_CACHE_PROVIDER_NAME);
+ + FILE_QUEUE_PROVIDER_NAME);
if (logFolderProp == null || logFolderProp.isEmpty()) {
logger.fatal("Audit spool folder is not configured. Please set "
+ propPrefix
+ "."
+ PROP_FILE_SPOOL_LOCAL_DIR
- + ". queueName=" + FILE_CACHE_PROVIDER_NAME);
+ + ". queueName=" + FILE_QUEUE_PROVIDER_NAME);
return false;
}
logFolder = new File(logFolderProp);
@@ -165,19 +180,19 @@ public class AuditFileCacheProviderSpool implements Runnable {
logger.fatal("File Spool folder not found and can't be created. folder="
+ logFolder.getAbsolutePath()
+ ", queueName="
- + FILE_CACHE_PROVIDER_NAME);
+ + FILE_QUEUE_PROVIDER_NAME);
return false;
}
}
logger.info("logFolder=" + logFolder + ", queueName="
- + FILE_CACHE_PROVIDER_NAME);
+ + FILE_QUEUE_PROVIDER_NAME);
if (logFileNameFormat == null || logFileNameFormat.isEmpty()) {
logFileNameFormat = "spool_" + "%app-type%" + "_"
+ "%time:yyyyMMdd-HHmm.ss%.log";
}
logger.info("logFileNameFormat=" + logFileNameFormat
- + ", queueName=" + FILE_CACHE_PROVIDER_NAME);
+ + ", queueName=" + FILE_QUEUE_PROVIDER_NAME);
if (archiveFolderProp == null || archiveFolderProp.isEmpty()) {
archiveFolder = new File(logFolder, "archive");
@@ -190,16 +205,16 @@ public class AuditFileCacheProviderSpool implements Runnable {
logger.error("File Spool archive folder not found and can't be created. folder="
+ archiveFolder.getAbsolutePath()
+ ", queueName="
- + FILE_CACHE_PROVIDER_NAME);
+ + FILE_QUEUE_PROVIDER_NAME);
return false;
}
}
logger.info("archiveFolder=" + archiveFolder + ", queueName="
- + FILE_CACHE_PROVIDER_NAME);
+ + FILE_QUEUE_PROVIDER_NAME);
if (indexFileName == null || indexFileName.isEmpty()) {
if (fileNamePrefix == null || fileNamePrefix.isEmpty()) {
- fileNamePrefix = FILE_CACHE_PROVIDER_NAME + "_"
+ fileNamePrefix = FILE_QUEUE_PROVIDER_NAME + "_"
+ consumerProvider.getName();
}
indexFileName = "index_" + fileNamePrefix + "_" + "%app-type%"
@@ -218,7 +233,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
}
}
logger.info("indexFile=" + indexFile + ", queueName="
- + FILE_CACHE_PROVIDER_NAME);
+ + FILE_QUEUE_PROVIDER_NAME);
int lastDot = indexFileName.lastIndexOf('.');
if (lastDot < 0) {
@@ -236,7 +251,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
}
}
logger.info("indexDoneFile=" + indexDoneFile + ", queueName="
- + FILE_CACHE_PROVIDER_NAME);
+ + FILE_QUEUE_PROVIDER_NAME);
// Load index file
loadIndexFile();
@@ -249,7 +264,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
currentWriterIndexRecord = auditIndexRecord;
logger.info("currentWriterIndexRecord="
+ currentWriterIndexRecord.filePath
- + ", queueName=" + FILE_CACHE_PROVIDER_NAME);
+ + ", queueName=" + FILE_QUEUE_PROVIDER_NAME);
}
if (auditIndexRecord.status
.equals(SPOOL_FILE_STATUS.read_inprogress)) {
@@ -270,14 +285,23 @@ public class AuditFileCacheProviderSpool implements Runnable {
}
}
+ auditFileType = MiscUtil.getStringProperty(props, propPrefix + ".filetype", DEFAULT_AUDIT_FILE_TYPE);
+ if (auditFileType == null) {
+ auditFileType = DEFAULT_AUDIT_FILE_TYPE;
+ }
+
} catch (Throwable t) {
logger.fatal("Error initializing File Spooler. queue="
- + FILE_CACHE_PROVIDER_NAME, t);
+ + FILE_QUEUE_PROVIDER_NAME, t);
return false;
}
+
+ bufferSize = MiscUtil.getLongProperty(props, propPrefix
+ + "." + PROP_FILE_SPOOL_BATCH_SIZE, bufferSize);
+
initDone = true;
- logger.debug("<== AuditFileCacheProviderSpool.init()");
+ logger.debug("<== AuditFileQueueSpool.init()");
return true;
}
@@ -287,16 +311,16 @@ public class AuditFileCacheProviderSpool implements Runnable {
public void start() {
if (!initDone) {
logger.error("Cannot start Audit File Spooler. Initilization not done yet. queueName="
- + FILE_CACHE_PROVIDER_NAME);
+ + FILE_QUEUE_PROVIDER_NAME);
return;
}
logger.info("Starting writerThread, queueName="
- + FILE_CACHE_PROVIDER_NAME + ", consumer="
+ + FILE_QUEUE_PROVIDER_NAME + ", consumer="
+ consumerProvider.getName());
// Let's start the thread to read
- destinationThread = new Thread(this, FILE_CACHE_PROVIDER_NAME + "_"
+ destinationThread = new Thread(this, FILE_QUEUE_PROVIDER_NAME + "_"
+ consumerProvider.getName() + "_destWriter");
destinationThread.setDaemon(true);
destinationThread.start();
@@ -305,10 +329,10 @@ public class AuditFileCacheProviderSpool implements Runnable {
public void stop() {
if (!initDone) {
logger.error("Cannot stop Audit File Spooler. Initilization not done. queueName="
- + FILE_CACHE_PROVIDER_NAME);
+ + FILE_QUEUE_PROVIDER_NAME);
return;
}
- logger.info("Stop called, queueName=" + FILE_CACHE_PROVIDER_NAME
+ logger.info("Stop called, queueName=" + FILE_QUEUE_PROVIDER_NAME
+ ", consumer=" + consumerProvider.getName());
isDrain = true;
@@ -329,7 +353,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
}
try {
logger.info("Closing open file, queueName="
- + FILE_CACHE_PROVIDER_NAME + ", consumer="
+ + FILE_QUEUE_PROVIDER_NAME + ", consumer="
+ consumerProvider.getName());
out.flush();
@@ -353,7 +377,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
public void flush() {
if (!initDone) {
logger.error("Cannot flush Audit File Spooler. Initilization not done. queueName="
- + FILE_CACHE_PROVIDER_NAME);
+ + FILE_QUEUE_PROVIDER_NAME);
return;
}
PrintWriter out = getOpenLogFileStream();
@@ -371,7 +395,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
public boolean isPending() {
if (!initDone) {
logError("isPending(): File Spooler not initialized. queueName="
- + FILE_CACHE_PROVIDER_NAME);
+ + FILE_QUEUE_PROVIDER_NAME);
return false;
}
@@ -494,7 +518,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
}
fileName = newFileName;
logger.info("Creating new file. queueName="
- + FILE_CACHE_PROVIDER_NAME + ", fileName=" + fileName);
+ + FILE_QUEUE_PROVIDER_NAME + ", fileName=" + fileName);
// Open the file
logWriter = new PrintWriter(new BufferedWriter(new OutputStreamWriter(new FileOutputStream(
outLogFile),"UTF-8")));
@@ -515,7 +539,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
// This means the process just started. We need to open the file
// in append mode.
logger.info("Opening existing file for append. queueName="
- + FILE_CACHE_PROVIDER_NAME + ", fileName="
+ + FILE_QUEUE_PROVIDER_NAME + ", fileName="
+ currentWriterIndexRecord.filePath);
logWriter = new PrintWriter(new BufferedWriter(new OutputStreamWriter(new FileOutputStream(
currentWriterIndexRecord.filePath, true),"UTF-8")));
@@ -544,7 +568,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
currentWriterIndexRecord.writeCompleteTime = new Date();
saveIndexFile();
logger.info("Adding file to queue. queueName="
- + FILE_CACHE_PROVIDER_NAME + ", fileName="
+ + FILE_QUEUE_PROVIDER_NAME + ", fileName="
+ currentWriterIndexRecord.filePath);
indexQueue.add(currentWriterIndexRecord);
currentWriterIndexRecord = null;
@@ -557,7 +581,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
- currentWriterIndexRecord.fileCreateTime.getTime() > fileRolloverSec * 1000) {
closeFile = true;
logger.info("Closing file. Rolling over. queueName="
- + FILE_CACHE_PROVIDER_NAME + ", fileName="
+ + FILE_QUEUE_PROVIDER_NAME + ", fileName="
+ currentWriterIndexRecord.filePath);
}
}
@@ -571,7 +595,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
logger.info("Loading index file. fileName=" + indexFile.getPath());
BufferedReader br = null;
try {
- br = new BufferedReader(new InputStreamReader(new FileInputStream(indexFile), "UTF-8"));
+ br = new BufferedReader(new InputStreamReader(new FileInputStream(indexFile), "UTF-8"));
indexRecords.clear();
String line;
while ((line = br.readLine()) != null) {
@@ -606,7 +630,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
AuditIndexRecord record = iter.next();
if (record.id.equals(indexRecord.id)) {
logger.info("Removing file from index. file=" + record.filePath
- + ", queueName=" + FILE_CACHE_PROVIDER_NAME
+ + ", queueName=" + FILE_QUEUE_PROVIDER_NAME
+ ", consumer=" + consumerProvider.getName());
iter.remove();
@@ -634,7 +658,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
void appendToDoneFile(AuditIndexRecord indexRecord)
throws FileNotFoundException, IOException {
logger.info("Moving to done file. " + indexRecord.filePath
- + ", queueName=" + FILE_CACHE_PROVIDER_NAME + ", consumer="
+ + ", queueName=" + FILE_QUEUE_PROVIDER_NAME + ", consumer="
+ consumerProvider.getName());
String line = gson.toJson(indexRecord);
PrintWriter out = new PrintWriter(new BufferedWriter(new OutputStreamWriter(new FileOutputStream(
@@ -778,7 +802,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
logger.info("Destination is down. sleeping for "
+ retryDestinationMS
+ " milli seconds. indexQueue=" + indexQueue.size()
- + ", queueName=" + FILE_CACHE_PROVIDER_NAME
+ + ", queueName=" + FILE_QUEUE_PROVIDER_NAME
+ ", consumer=" + consumerProvider.getName());
Thread.sleep(retryDestinationMS);
}
@@ -812,38 +836,19 @@ public class AuditFileCacheProviderSpool implements Runnable {
BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(
currentConsumerIndexRecord.filePath),"UTF-8"));
try {
- int startLine = currentConsumerIndexRecord.linePosition;
- String line;
- int currLine = 0;
- List<AuditEventBase> events = new ArrayList<>();
- while ((line = br.readLine()) != null) {
- currLine++;
- if (currLine < startLine) {
- continue;
- }
- AuditEventBase event = MiscUtil.fromJson(line, AuthzAuditEvent.class);
- events.add(event);
-
- if (events.size() == AUDIT_BATCH_SIZE_DEFAULT) {
- boolean ret = sendEvent(events,
- currentConsumerIndexRecord, currLine);
- if (!ret) {
- throw new Exception("Destination down");
- }
- events.clear();
- }
- }
- if (events.size() > 0) {
- boolean ret = sendEvent(events,
- currentConsumerIndexRecord, currLine);
- if (!ret) {
- throw new Exception("Destination down");
- }
- events.clear();
+ if (auditFileType.equalsIgnoreCase(DEFAULT_AUDIT_FILE_TYPE)) {
+ // if Audit File format is JSON each audit file in the Local Spool Location will be copied
+ // to HDFS location as JSON
+ File srcFile = new File(currentConsumerIndexRecord.filePath);
+ logFile(srcFile);
+ } else {
+ // If Audit File format is ORC, each records in audit files in the Local Spool Location will be
+ // read and converted into ORC format and pushed into an ORC file.
+ logEvent(br);
}
logger.info("Done reading file. file="
+ currentConsumerIndexRecord.filePath
- + ", queueName=" + FILE_CACHE_PROVIDER_NAME
+ + ", queueName=" + FILE_QUEUE_PROVIDER_NAME
+ ", consumer=" + consumerProvider.getName());
// The entire file is read
currentConsumerIndexRecord.status = SPOOL_FILE_STATUS.done;
@@ -854,7 +859,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
} catch (Exception ex) {
isDestDown = true;
logError("Destination down. queueName="
- + FILE_CACHE_PROVIDER_NAME + ", consumer="
+ + FILE_QUEUE_PROVIDER_NAME + ", consumer="
+ consumerProvider.getName());
lastAttemptTime = System.currentTimeMillis();
// Update the index file
@@ -878,10 +883,42 @@ public class AuditFileCacheProviderSpool implements Runnable {
logger.error("Exception in destination writing thread.", t);
}
}
- logger.info("Exiting file spooler. provider=" + FILE_CACHE_PROVIDER_NAME
+ logger.info("Exiting file spooler. provider=" + FILE_QUEUE_PROVIDER_NAME
+ ", consumer=" + consumerProvider.getName());
}
+ private void logEvent(BufferedReader br) throws Exception {
+ String line;
+ int currLine = 0;
+ int startLine = currentConsumerIndexRecord.linePosition;
+ List<AuditEventBase> events = new ArrayList<>();
+ while ((line = br.readLine()) != null) {
+ currLine++;
+ if (currLine < startLine) {
+ continue;
+ }
+ AuditEventBase event = MiscUtil.fromJson(line, AuthzAuditEvent.class);
+ events.add(event);
+
+ if (events.size() == bufferSize) {
+ boolean ret = sendEvent(events,
+ currentConsumerIndexRecord, currLine);
+ if (!ret) {
+ throw new Exception("Destination down");
+ }
+ events.clear();
+ }
+ }
+ if (events.size() > 0) {
+ boolean ret = sendEvent(events,
+ currentConsumerIndexRecord, currLine);
+ if (!ret) {
+ throw new Exception("Destination down");
+ }
+ events.clear();
+ }
+ }
+
private boolean sendEvent(List<AuditEventBase> events, AuditIndexRecord indexRecord,
int currLine) {
boolean ret = true;
@@ -890,7 +927,7 @@ public class AuditFileCacheProviderSpool implements Runnable {
if (!ret) {
// Need to log error after fixed interval
logError("Error sending logs to consumer. provider="
- + FILE_CACHE_PROVIDER_NAME + ", consumer="
+ + FILE_QUEUE_PROVIDER_NAME + ", consumer="
+ consumerProvider.getName());
} else {
// Update index and save
@@ -903,17 +940,78 @@ public class AuditFileCacheProviderSpool implements Runnable {
if (isDestDown) {
isDestDown = false;
logger.info("Destination up now. " + indexRecord.filePath
- + ", queueName=" + FILE_CACHE_PROVIDER_NAME
+ + ", queueName=" + FILE_QUEUE_PROVIDER_NAME
+ ", consumer=" + consumerProvider.getName());
}
}
} catch (Throwable t) {
logger.error("Error while sending logs to consumer. provider="
- + FILE_CACHE_PROVIDER_NAME + ", consumer="
+ + FILE_QUEUE_PROVIDER_NAME + ", consumer="
+ consumerProvider.getName() + ", log=" + events, t);
}
return ret;
}
-}
+ private void logFile(File file) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> AuditFileQueueSpool.logFile()");
+ }
+ int currLine = 0;
+ int startLine = currentConsumerIndexRecord.linePosition;
+
+ if (currLine < startLine) {
+ currLine++;
+ }
+
+ boolean ret = sendFile(file,currentConsumerIndexRecord, currLine);
+ if (!ret) {
+ throw new Exception("Destination down");
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== AuditFileQueueSpool.logFile()");
+ }
+ }
+
+ private boolean sendFile(File file, AuditIndexRecord indexRecord,
+ int currLine) {
+ boolean ret = true;
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> AuditFileQueueSpool.sendFile()");
+ }
+
+ try {
+ ret = consumerProvider.logFile(file);
+ if (!ret) {
+ // Need to log error after fixed interval
+ logError("Error sending log file to consumer. provider="
+ + FILE_QUEUE_PROVIDER_NAME + ", consumer="
+ + consumerProvider.getName()+ ", logFile=" + file.getName());
+ } else {
+ // Update index and save
+ indexRecord.linePosition = currLine;
+ indexRecord.status = SPOOL_FILE_STATUS.read_inprogress;
+ indexRecord.lastSuccessTime = new Date();
+ indexRecord.lastAttempt = true;
+ saveIndexFile();
+
+ if (isDestDown) {
+ isDestDown = false;
+ logger.info("Destination up now. " + indexRecord.filePath
+ + ", queueName=" + FILE_QUEUE_PROVIDER_NAME
+ + ", consumer=" + consumerProvider.getName());
+ }
+ }
+ } catch (Throwable t) {
+ logger.error("Error sending log file to consumer. provider="
+ + FILE_QUEUE_PROVIDER_NAME + ", consumer="
+ + consumerProvider.getName() + ", logFile=" + file.getName(), t);
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== AuditFileQueueSpool.sendFile() " + ret );
+ }
+ return ret;
+ }
+}
\ No newline at end of file
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/utils/AbstractRangerAuditWriter.java b/agents-audit/src/main/java/org/apache/ranger/audit/utils/AbstractRangerAuditWriter.java
new file mode 100644
index 0000000..b5967e4
--- /dev/null
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/utils/AbstractRangerAuditWriter.java
@@ -0,0 +1,353 @@
+package org.apache.ranger.audit.utils;
+
+/*
+ * 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.
+ */
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.*;
+import org.apache.ranger.audit.provider.MiscUtil;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.PrintWriter;
+import java.net.URI;
+import java.util.Date;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * This is Abstract class to have common properties of Ranger Audit HDFS Destination Writer.
+ */
+public abstract class AbstractRangerAuditWriter implements RangerAuditWriter {
+ private static final Log logger = LogFactory.getLog(AbstractRangerAuditWriter.class);
+
+ public static final String PROP_FILESYSTEM_DIR = "dir";
+ public static final String PROP_FILESYSTEM_SUBDIR = "subdir";
+ public static final String PROP_FILESYSTEM_FILE_NAME_FORMAT = "filename.format";
+ public static final String PROP_FILESYSTEM_FILE_ROLLOVER = "file.rollover.sec";
+ public static final String PROP_FILESYSTEM_ROLLOVER_PERIOD = "file.rollover.period";
+ public static final String PROP_FILESYSTEM_FILE_EXTENSION = ".log";
+ public Configuration conf = null;
+ public FileSystem fileSystem = null;
+ public Map<String, String> auditConfigs = null;
+ public Path auditPath = null;
+ public PrintWriter logWriter = null;
+ public RollingTimeUtil rollingTimeUtil = null;
+ public String auditProviderName = null;
+ public String fullPath = null;
+ public String parentFolder = null;
+ public String currentFileName = null;
+ public String logFileNameFormat = null;
+ public String logFolder = null;
+ public String fileExtension = null;
+ public String rolloverPeriod = null;
+ public String fileSystemScheme = null;
+ public Date nextRollOverTime = null;
+ public int fileRolloverSec = 24 * 60 * 60; // In seconds
+ public boolean rollOverByDuration = false;
+ public volatile FSDataOutputStream ostream = null; // output stream wrapped in logWriter
+ private boolean isHFlushCapableStream = false;
+
+ @Override
+ public void init(Properties props, String propPrefix, String auditProviderName, Map<String,String> auditConfigs) {
+ // Initialize properties for this class
+ // Initial folder and file properties
+ logger.info("==> AbstractRangerAuditWriter.init()");
+ this.auditProviderName = auditProviderName;
+ this.auditConfigs = auditConfigs;
+
+ init(props,propPrefix);
+
+ logger.info("<== AbstractRangerAuditWriter.init()");
+ }
+
+ public void createFileSystemFolders() throws Exception {
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> AbstractRangerAuditWriter.createFileSystemFolders()");
+ }
+ // Create a new file
+ Date currentTime = new Date();
+ String fileName = MiscUtil.replaceTokens(logFileNameFormat, currentTime.getTime());
+ parentFolder = MiscUtil.replaceTokens(logFolder, currentTime.getTime());
+ fullPath = parentFolder + Path.SEPARATOR + fileName;
+ String defaultPath = fullPath;
+ conf = createConfiguration();
+ URI uri = URI.create(fullPath);
+ fileSystem = FileSystem.get(uri, conf);
+ auditPath = new Path(fullPath);
+ fileSystemScheme = getFileSystemScheme();
+ logger.info("Checking whether log file exists. "+ fileSystemScheme + "Path= " + fullPath + ", UGI=" + MiscUtil.getUGILoginUser());
+ int i = 0;
+ while (fileSystem.exists(auditPath)) {
+ i++;
+ int lastDot = defaultPath.lastIndexOf('.');
+ String baseName = defaultPath.substring(0, lastDot);
+ String extension = defaultPath.substring(lastDot);
+ fullPath = baseName + "." + i + extension;
+ auditPath = new Path(fullPath);
+ logger.info("Checking whether log file exists. "+ fileSystemScheme + "Path= " + fullPath);
+ }
+ logger.info("Log file doesn't exists. Will create and use it. "+ fileSystemScheme + "Path= " + fullPath);
+
+ // Create parent folders
+ createParents(auditPath, fileSystem);
+
+ currentFileName = fullPath;
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== AbstractRangerAuditWriter.createFileSystemFolders()");
+ }
+ }
+
+ public Configuration createConfiguration() {
+ Configuration conf = new Configuration();
+ for (Map.Entry<String, String> entry : auditConfigs.entrySet()) {
+ String key = entry.getKey();
+ String value = entry.getValue();
+ // for ease of install config file may contain properties with empty value, skip those
+ if (StringUtils.isNotEmpty(value)) {
+ conf.set(key, value);
+ }
+ logger.info("Adding property to "+ fileSystemScheme + " + config: " + key + " => " + value);
+ }
+
+ logger.info("Returning " + fileSystemScheme + "Filesystem Config: " + conf.toString());
+ return conf;
+ }
+
+ public void createParents(Path pathLogfile, FileSystem fileSystem)
+ throws Exception {
+ logger.info("Creating parent folder for " + pathLogfile);
+ Path parentPath = pathLogfile != null ? pathLogfile.getParent() : null;
+
+ if (parentPath != null && fileSystem != null
+ && !fileSystem.exists(parentPath)) {
+ fileSystem.mkdirs(parentPath);
+ }
+ }
+
+ public void init(Properties props, String propPrefix) {
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> AbstractRangerAuditWriter.init()");
+ }
+
+ String logFolderProp = MiscUtil.getStringProperty(props, propPrefix + "." + PROP_FILESYSTEM_DIR);
+ if (StringUtils.isEmpty(logFolderProp)) {
+ logger.fatal("File destination folder is not configured. Please set "
+ + propPrefix + "."
+ + PROP_FILESYSTEM_DIR + ". name="
+ + auditProviderName);
+ return;
+ }
+
+ String logSubFolder = MiscUtil.getStringProperty(props, propPrefix + "." + PROP_FILESYSTEM_SUBDIR);
+ if (StringUtils.isEmpty(logSubFolder)) {
+ logSubFolder = "%app-type%/%time:yyyyMMdd%";
+ }
+
+ logFileNameFormat = MiscUtil.getStringProperty(props, propPrefix + "." + PROP_FILESYSTEM_FILE_NAME_FORMAT);
+ fileRolloverSec = MiscUtil.getIntProperty(props, propPrefix + "." + PROP_FILESYSTEM_FILE_ROLLOVER, fileRolloverSec);
+
+ if (StringUtils.isEmpty(fileExtension)) {
+ setFileExtension(PROP_FILESYSTEM_FILE_EXTENSION);
+ }
+
+ if (logFileNameFormat == null || logFileNameFormat.isEmpty()) {
+ logFileNameFormat = "%app-type%_ranger_audit_%hostname%" + fileExtension;
+ }
+
+ logFolder = logFolderProp + "/" + logSubFolder;
+
+ logger.info("logFolder=" + logFolder + ", destName=" + auditProviderName);
+ logger.info("logFileNameFormat=" + logFileNameFormat + ", destName="+ auditProviderName);
+ logger.info("config=" + auditConfigs.toString());
+
+ rolloverPeriod = MiscUtil.getStringProperty(props, propPrefix + "." + PROP_FILESYSTEM_ROLLOVER_PERIOD);
+ rollingTimeUtil = RollingTimeUtil.getInstance();
+
+ //file.rollover.period is used for rolling over. If it could compute the next roll over time using file.rollover.period
+ //it fall back to use file.rollover.sec for find next rollover time. If still couldn't find default will be 1day window
+ //for rollover.
+ if(StringUtils.isEmpty(rolloverPeriod) ) {
+ rolloverPeriod = rollingTimeUtil.convertRolloverSecondsToRolloverPeriod(fileRolloverSec);
+ }
+
+ try {
+ nextRollOverTime = rollingTimeUtil.computeNextRollingTime(rolloverPeriod);
+ } catch ( Exception e) {
+ logger.warn("Rollover by file.rollover.period failed...will be using the file.rollover.sec for "+ fileSystemScheme + " audit file rollover...", e);
+ rollOverByDuration = true;
+ nextRollOverTime = rollOverByDuration();
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== AbstractRangerAuditWriter.init()");
+ }
+
+ }
+
+ public void closeFileIfNeeded() throws IOException {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> AbstractRangerAuditWriter.closeFileIfNeeded()");
+ }
+
+ if (logWriter == null) {
+ return;
+ }
+
+ if ( System.currentTimeMillis() > nextRollOverTime.getTime() ) {
+ logger.info("Closing file. Rolling over. name=" + auditProviderName
+ + ", fileName=" + currentFileName);
+ try {
+ logWriter.flush();
+ logWriter.close();
+ } catch (Throwable t) {
+ logger.error("Error on closing log writter. Exception will be ignored. name="
+ + auditProviderName + ", fileName=" + currentFileName);
+ }
+
+ logWriter = null;
+ ostream = null;
+ currentFileName = null;
+
+ if (!rollOverByDuration) {
+ try {
+ if(StringUtils.isEmpty(rolloverPeriod) ) {
+ rolloverPeriod = rollingTimeUtil.convertRolloverSecondsToRolloverPeriod(fileRolloverSec);
+ }
+ nextRollOverTime = rollingTimeUtil.computeNextRollingTime(rolloverPeriod);
+ } catch ( Exception e) {
+ logger.warn("Rollover by file.rollover.period failed...will be using the file.rollover.sec for " + fileSystemScheme + " audit file rollover...", e);
+ nextRollOverTime = rollOverByDuration();
+ }
+ } else {
+ nextRollOverTime = rollOverByDuration();
+ }
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== AbstractRangerAuditWriter.closeFileIfNeeded()");
+ }
+ }
+
+ public Date rollOverByDuration() {
+ long rollOverTime = rollingTimeUtil.computeNextRollingTime(fileRolloverSec,nextRollOverTime);
+ return new Date(rollOverTime);
+ }
+
+ public PrintWriter createWriter() throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> AbstractRangerAuditWriter.createWriter()");
+ }
+
+ if (logWriter == null) {
+ // Create the file to write
+ logger.info("Creating new log file. auditPath=" + fullPath);
+ createFileSystemFolders();
+ ostream = fileSystem.create(auditPath);
+ logWriter = new PrintWriter(ostream);
+ isHFlushCapableStream = ostream.hasCapability(StreamCapabilities.HFLUSH);
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== AbstractRangerAuditWriter.createWriter()");
+ }
+
+ return logWriter;
+ }
+
+ public void closeWriter() {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> AbstractRangerAuditWriter.closeWriter()");
+ }
+
+ logWriter = null;
+ ostream = null;
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== AbstractRangerAuditWriter.closeWriter()");
+ }
+ }
+
+ @Override
+ public void flush() {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> AbstractRangerAuditWriter.flush()");
+ }
+ if (ostream != null) {
+ try {
+ synchronized (this) {
+ if (ostream != null)
+ // 1) PrinterWriter does not have bufferring of its own so
+ // we need to flush its underlying stream
+ // 2) HDFS flush() does not really flush all the way to disk.
+ if (isHFlushCapableStream) {
+ //Checking HFLUSH capability of the stream because of HADOOP-13327.
+ //For S3 filesysttem, hflush throws UnsupportedOperationException and hence we call flush.
+ ostream.hflush();
+ } else {
+ ostream.flush();
+ } logger.info("Flush " + fileSystemScheme + " audit logs completed.....");
+ }
+ } catch (IOException e) {
+ logger.error("Error on flushing log writer: " + e.getMessage() +
+ "\nException will be ignored. name=" + auditProviderName + ", fileName=" + currentFileName);
+ }
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== AbstractRangerAuditWriter.flush()");
+ }
+ }
+
+ public boolean logFileToHDFS(File file) throws Exception {
+ boolean ret = false;
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> AbstractRangerAuditWriter.logFileToHDFS()");
+ }
+
+ if (logWriter == null) {
+ // Create the file to write
+ createFileSystemFolders();
+ logger.info("Copying the Audit File" + file.getName() + " to HDFS Path" + fullPath);
+ Path destPath = new Path(fullPath);
+ ret = FileUtil.copy(file,fileSystem,destPath,false,conf);
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== AbstractRangerAuditWriter.logFileToHDFS()");
+ }
+ return ret;
+ }
+
+ public String getFileSystemScheme() {
+ String ret = null;
+ ret = logFolder.substring(0, (logFolder.indexOf(":")));
+ ret = ret.toUpperCase();
+ return ret;
+ }
+
+ public void setFileExtension(String fileExtension) {
+ this.fileExtension = fileExtension;
+ }
+}
\ No newline at end of file
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/utils/ORCFileUtil.java b/agents-audit/src/main/java/org/apache/ranger/audit/utils/ORCFileUtil.java
new file mode 100644
index 0000000..2e4378d
--- /dev/null
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/utils/ORCFileUtil.java
@@ -0,0 +1,454 @@
+package org.apache.ranger.audit.utils;
+
+/*
+ * 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.
+ */
+
+import com.google.gson.annotations.SerializedName;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.exec.vector.*;
+import org.apache.orc.CompressionKind;
+import org.apache.orc.OrcFile;
+import org.apache.orc.OrcFile.WriterOptions;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.Writer;
+import org.apache.ranger.audit.model.AuthzAuditEvent;
+import org.apache.ranger.audit.model.EnumRepositoryType;
+
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Date;
+import java.util.Map;
+import java.util.HashMap;
+import java.text.Format;
+import java.text.SimpleDateFormat;
+
+public class ORCFileUtil {
+
+ private static final Log logger = LogFactory.getLog(ORCFileUtil.class);
+
+ private static volatile ORCFileUtil me = null;
+ protected CompressionKind defaultCompression = CompressionKind.SNAPPY;
+ protected CompressionKind compressionKind = CompressionKind.NONE;
+ protected TypeDescription schema = null;
+ protected VectorizedRowBatch batch = null;
+ protected String auditSchema = null;
+ protected String dateFormat = "yyyy-MM-dd HH:mm:ss";
+
+ protected ArrayList<String> schemaFields = new ArrayList<>();
+ protected Map<String,ColumnVector> vectorizedRowBatchMap = new HashMap<>();
+ protected int orcBufferSize;
+ protected long orcStripeSize;
+
+ public static ORCFileUtil getInstance() {
+ ORCFileUtil orcFileUtil = me;
+ if (orcFileUtil == null) {
+ synchronized (ORCFileUtil.class) {
+ orcFileUtil = me;
+ if (orcFileUtil == null) {
+ me = orcFileUtil = new ORCFileUtil();
+ }
+ }
+ }
+ return orcFileUtil;
+ }
+
+ public void init(int orcBufferSize, long orcStripeSize, String compression) throws Exception{
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> ORCFileUtil.init()");
+ }
+ this.orcBufferSize = orcBufferSize;
+ this.orcStripeSize = orcStripeSize;
+ this.compressionKind = getORCCompression(compression);
+ initORCAuditSchema();
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== ORCFileUtil.init() : orcBufferSize: " + orcBufferSize + " stripeSize: " + orcStripeSize +
+ " compression: " + compression);
+ }
+ }
+
+ public Writer createWriter(Configuration conf, FileSystem fs, String path) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> ORCFileUtil.createWriter()");
+ }
+ Writer ret = null;
+ WriterOptions writeOptions = OrcFile.writerOptions(conf)
+ .fileSystem(fs)
+ .setSchema(schema)
+ .bufferSize(orcBufferSize)
+ .stripeSize(orcStripeSize)
+ .compress(compressionKind);
+
+ ret = OrcFile.createWriter(new Path(path), writeOptions);
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== ORCFileUtil.createWriter()");
+ }
+ return ret;
+ }
+
+ public void close(Writer writer) throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> ORCFileUtil.close()");
+ }
+
+ writer.close();
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== ORCFileUtil.close()");
+ }
+ }
+
+ public void log(Writer writer, Collection<AuthzAuditEvent> events) throws Exception {
+ int eventBatchSize = events.size();
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> ORCFileUtil.log() : EventSize: " + eventBatchSize + "ORC bufferSize:" + orcBufferSize );
+ }
+
+ //increase the batch size according to event size, so it can accomodate all the events.
+ if (eventBatchSize > orcBufferSize) {
+ batch = schema.createRowBatch(orcBufferSize);
+ }
+
+ try {
+ for(AuthzAuditEvent event : events) {
+ int row = batch.size++;
+ for (int j=0;j<schemaFields.size();j++) {
+ String fieldName = schemaFields.get(j);
+ SchemaInfo schemaInfo = getFieldValue(event, fieldName);
+ ColumnVector columnVector = vectorizedRowBatchMap.get(fieldName);
+ if (columnVector instanceof LongColumnVector) {
+ ((LongColumnVector) columnVector).vector[row] = castLongObject(schemaInfo.getValue());
+ } else if (columnVector instanceof BytesColumnVector) {
+ ((BytesColumnVector) columnVector).setVal(row, getBytesValues(castStringObject(schemaInfo.getValue())));
+ }
+ }
+ if (batch.size == orcBufferSize) {
+ writer.addRowBatch(batch);
+ batch.reset();
+ }
+ }
+ if (batch.size != 0) {
+ writer.addRowBatch(batch);
+ batch.reset();
+ }
+ } catch (Exception e) {
+ batch.reset();
+ logger.error("Error while writing into ORC File:", e);
+ throw e;
+ }
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== ORCFileUtil.log(): EventSize = " + eventBatchSize );
+ }
+ }
+
+ protected byte[] getBytesValues(String val) {
+ byte[] ret = "".getBytes();
+ if(val != null) {
+ ret = val.getBytes();
+ }
+ return ret;
+ }
+
+ protected String getDateString(Date date) {
+ String ret = null;
+ Format formatter = new SimpleDateFormat(dateFormat);
+ ret = formatter.format(date);
+ return ret;
+ }
+
+ protected void initORCAuditSchema() throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> ORCWriter.initORCAuditSchema()");
+ }
+ auditSchema = getAuditSchema();
+ Map<String,String> schemaFieldTypeMap = getSchemaFieldTypeMap();
+ schema = TypeDescription.fromString(auditSchema);
+ batch = schema.createRowBatch(orcBufferSize);
+ buildVectorRowBatch(schemaFieldTypeMap);
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== ORCWriter.initORCAuditSchema()");
+ }
+ }
+
+ protected Map<String,String> getSchemaFieldTypeMap() {
+ Map<String,String> ret = new HashMap<>();
+
+ int index1 = auditSchema.indexOf("<");
+ int index2 = auditSchema.indexOf(">");
+ String subAuditSchema = auditSchema.substring(index1+1,index2);
+ String[] fields = subAuditSchema.split(",");
+ schemaFields = new ArrayList<>();
+
+ for (String field: fields) {
+ String[] flds = field.split(":");
+ schemaFields.add(flds[0]);
+ ret.put(flds[0],flds[1]);
+ }
+ return ret;
+ }
+
+ protected void buildVectorRowBatch(Map<String,String> schemaFieldTypeMap) throws Exception {
+ int i = 0;
+ for (i=0;i<schemaFields.size();i++) {
+ String fld = schemaFields.get(i);
+ String fieldType = schemaFieldTypeMap.get(fld);
+ ColumnVector columnVector = getColumnVectorType(fieldType);
+ if (columnVector instanceof LongColumnVector) {
+ vectorizedRowBatchMap.put(fld, (LongColumnVector) batch.cols[i]);
+ } else if (columnVector instanceof BytesColumnVector) {
+ vectorizedRowBatchMap.put(fld, (BytesColumnVector) batch.cols[i]);
+ } else if (columnVector instanceof DecimalColumnVector) {
+ vectorizedRowBatchMap.put(fld, (DecimalColumnVector) batch.cols[i]);
+ }
+ }
+ }
+
+ protected SchemaInfo getFieldValue(AuthzAuditEvent event, String fieldName ) {
+ SchemaInfo ret = new SchemaInfo();
+ try {
+ Class aClass = AuthzAuditEvent.class;
+ Field fld = aClass.getDeclaredField(fieldName);
+ fld.setAccessible(true);
+
+ Class<?> cls = fld.getType();
+ Object value = fld.get(event);
+
+ ret.setField(fieldName);
+ ret.setType(cls.getName());
+ ret.setValue(value);
+ } catch (Exception e){
+ logger.error("Error while writing into ORC File:", e);
+ }
+ return ret;
+ }
+
+ protected ColumnVector getColumnVectorType(String fieldType) throws Exception {
+ ColumnVector ret = null;
+ fieldType = fieldType.toLowerCase();
+ switch(fieldType) {
+ case "int" :
+ case "bigint":
+ case "date":
+ case "boolean":
+ ret = new LongColumnVector();
+ break;
+ case "string":
+ case "varchar":
+ case "char":
+ case "binary":
+ ret = new BytesColumnVector();
+ break;
+ case "decimal":
+ ret = new DecimalColumnVector(10,5);
+ break;
+ case "double":
+ case "float":
+ ret = new DoubleColumnVector();
+ break;
+ case "array":
+ case "map":
+ case "uniontype":
+ case "struct":
+ throw new Exception("Unsuppoted field Type");
+ }
+ return ret;
+ }
+
+ protected Long castLongObject(Object object) {
+ Long ret = 0l;
+ try {
+ if (object instanceof Long)
+ ret = ((Long) object).longValue();
+ else if (object instanceof Integer) {
+ ret = ((Integer) object).longValue();
+ } else if (object instanceof String) {
+ ret = Long.valueOf((String) object);
+ }
+ } catch (Exception e) {
+ logger.error("Error while writing into ORC File:", e);
+ }
+ return ret;
+ }
+
+ protected String castStringObject(Object object) {
+ String ret = null;
+ try {
+ if (object instanceof String)
+ ret = (String) object;
+ else if (object instanceof Date) {
+ ret = (getDateString((Date) object));
+ }
+ } catch (Exception e) {
+ logger.error("Error while writing into ORC File:", e);
+ }
+ return ret;
+ }
+
+ protected String getAuditSchema() {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> ORCWriter.getAuditSchema()");
+ }
+ String ret = null;
+ String fieldStr = "struct<";
+ StringBuilder sb = new StringBuilder(fieldStr);
+
+ Class auditEventClass = AuthzAuditEvent.class;
+ for(Field fld: auditEventClass.getDeclaredFields()) {
+ if (fld.isAnnotationPresent(SerializedName.class)) {
+ String field = fld.getName();
+ String fieldType = getShortFieldType(fld.getType().getName());
+ if (fieldType == null) {
+ continue;
+ }
+ fieldStr = field + ":" + fieldType + ",";
+ sb.append(fieldStr);
+ }
+ }
+ fieldStr = sb.toString();
+ if (fieldStr.endsWith(",")) {
+ fieldStr = fieldStr.substring(0, fieldStr.length() - 1);
+ }
+ ret = fieldStr + ">";
+
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== ORCWriter.getAuditSchema() AuditSchema: " + ret);
+ }
+ return ret;
+ }
+
+ protected String getShortFieldType(String type){
+ String ret = null;
+ switch(type) {
+ case "java.lang.String":
+ ret = "string";
+ break;
+ case "int":
+ ret = "int";
+ break;
+ case "short":
+ ret = "string";
+ break;
+ case "java.util.Date":
+ ret = "string";
+ break;
+ case "long":
+ ret = "bigint";
+ break;
+ default:
+ ret = null;
+ }
+ return ret;
+ }
+
+ class SchemaInfo {
+ String field = null;
+ String type = null;
+ Object value = null;
+
+ public String getField() {
+ return field;
+ }
+
+ public void setField(String field) {
+ this.field = field;
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ public void setType(String type) {
+ this.type = type;
+ }
+
+ public Object getValue() {
+ return value;
+ }
+
+ public void setValue(Object value) {
+ this.value = value;
+ }
+ }
+
+ protected CompressionKind getORCCompression(String compression) {
+ CompressionKind ret;
+ if (compression == null) {
+ compression = defaultCompression.name().toLowerCase();
+ }
+ switch(compression) {
+ case "snappy":
+ ret = CompressionKind.SNAPPY;
+ break;
+ case "lzo":
+ ret = CompressionKind.LZO;
+ break;
+ case "zlip":
+ ret = CompressionKind.ZLIB;
+ break;
+ case "none":
+ ret = CompressionKind.NONE;
+ break;
+ default:
+ ret = defaultCompression;
+ break;
+ }
+ return ret;
+ }
+
+ public static void main(String[] args) throws Exception {
+ ORCFileUtil auditOrcFileUtil = new ORCFileUtil();
+ auditOrcFileUtil.init(10000,100000L,"snappy");
+ try {
+ Configuration conf = new Configuration();
+ FileSystem fs = FileSystem.get(conf);
+ Writer write = auditOrcFileUtil.createWriter(conf, fs, "/tmp/test.orc");
+ Collection<AuthzAuditEvent> events = getTestEvent();
+ auditOrcFileUtil.log(write, events);
+ write.close();
+ } catch (Exception e){
+ e.printStackTrace();
+ }
+ }
+
+ protected static Collection<AuthzAuditEvent> getTestEvent() {
+ Collection<AuthzAuditEvent> events = new ArrayList<>();
+ for (int idx=0;idx<20;idx++) {
+ AuthzAuditEvent event = new AuthzAuditEvent();
+ event.setEventId(Integer.toString(idx));
+ event.setClientIP("127.0.0.1");
+ event.setAccessResult((short) 1);
+ event.setAclEnforcer("ranger-acl");
+ event.setRepositoryName("hdfsdev");
+ event.setRepositoryType(EnumRepositoryType.HDFS);
+ event.setResourcePath("/tmp/test-audit.log" +idx+idx+1);
+ event.setResourceType("file");
+ event.setAccessType("read");
+ event.setEventTime(new Date());
+ event.setResultReason(Integer.toString(1));
+ events.add(event);
+ }
+ return events;
+ }
+}
\ No newline at end of file
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/utils/RangerAuditWriter.java b/agents-audit/src/main/java/org/apache/ranger/audit/utils/RangerAuditWriter.java
new file mode 100644
index 0000000..fbe9301
--- /dev/null
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/utils/RangerAuditWriter.java
@@ -0,0 +1,39 @@
+package org.apache.ranger.audit.utils;
+
+/*
+ * 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.
+ */
+
+import java.io.File;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+
+public interface RangerAuditWriter {
+ void init(Properties prop, String propPrefix, String auditProviderName, Map<String,String> auditConfigs);
+
+ boolean log(Collection<String> events) throws Exception;
+
+ boolean logFile(File file) throws Exception;
+
+ void start();
+
+ void flush();
+
+ void stop();
+}
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/utils/RangerJSONAuditWriter.java b/agents-audit/src/main/java/org/apache/ranger/audit/utils/RangerJSONAuditWriter.java
new file mode 100644
index 0000000..f46e0a5
--- /dev/null
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/utils/RangerJSONAuditWriter.java
@@ -0,0 +1,175 @@
+package org.apache.ranger.audit.utils;
+
+/*
+ * 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.
+ */
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.ranger.audit.provider.MiscUtil;
+
+import java.io.File;
+import java.io.PrintWriter;
+import java.security.PrivilegedExceptionAction;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Writes the Ranger audit to HDFS as JSON text
+ */
+public class RangerJSONAuditWriter extends AbstractRangerAuditWriter {
+
+ private static final Log logger = LogFactory.getLog(RangerJSONAuditWriter.class);
+
+ protected String JSON_FILE_EXTENSION = ".log";
+
+ public void init(Properties props, String propPrefix, String auditProviderName, Map<String,String> auditConfigs) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerJSONAuditWriter.init()");
+ }
+ init();
+ super.init(props,propPrefix,auditProviderName,auditConfigs);
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerJSONAuditWriter.init()");
+ }
+ }
+
+ public void init() {
+ setFileExtension(JSON_FILE_EXTENSION);
+ }
+
+ synchronized public boolean logJSON(final Collection<String> events) throws Exception {
+ boolean ret = false;
+ PrintWriter out = null;
+ try {
+ if (logger.isDebugEnabled()) {
+ logger.debug("UGI=" + MiscUtil.getUGILoginUser()
+ + ". Will write to HDFS file=" + currentFileName);
+ }
+ out = MiscUtil.executePrivilegedAction(new PrivilegedExceptionAction<PrintWriter>() {
+ @Override
+ public PrintWriter run() throws Exception {
+ PrintWriter out = getLogFileStream();
+ for (String event : events) {
+ out.println(event);
+ }
+ return out;
+ };
+ });
+ // flush and check the stream for errors
+ if (out.checkError()) {
+ // In theory, this count may NOT be accurate as part of the messages may have been successfully written.
+ // However, in practice, since client does buffering, either all of none would succeed.
+ out.close();
+ closeWriter();
+ return ret;
+ }
+ } catch (Exception e) {
+ if (out != null) {
+ out.close();
+ }
+ closeWriter();
+ return ret;
+ } finally {
+ ret = true;
+ if (logger.isDebugEnabled()) {
+ logger.debug("Flushing HDFS audit. Event Size:" + events.size());
+ }
+ if (out != null) {
+ out.flush();
+ }
+ }
+
+ return ret;
+ }
+
+ @Override
+ public boolean log(Collection<String> events) throws Exception {
+ return logJSON(events);
+ }
+
+ synchronized public boolean logAsFile(final File file) throws Exception {
+ boolean ret = false;
+ if (logger.isDebugEnabled()) {
+ logger.debug("UGI=" + MiscUtil.getUGILoginUser()
+ + ". Will write to HDFS file=" + currentFileName);
+ }
+ Boolean retVal = MiscUtil.executePrivilegedAction(new PrivilegedExceptionAction<Boolean>() {
+ @Override
+ public Boolean run() throws Exception {
+ boolean ret = logFileToHDFS(file);
+ return Boolean.valueOf(ret);
+ };
+ });
+ ret = retVal.booleanValue();
+ logger.info("Flushing HDFS audit File :" + file.getAbsolutePath() + file.getName());
+ return ret;
+ }
+
+ @Override
+ public boolean logFile(File file) throws Exception {
+ return logAsFile(file);
+ }
+
+ synchronized public PrintWriter getLogFileStream() throws Exception {
+ closeFileIfNeeded();
+ // Either there are no open log file or the previous one has been rolled
+ // over
+ PrintWriter logWriter = createWriter();
+ return logWriter;
+ }
+
+
+ public void flush() {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> JSONWriter.flush()");
+ }
+ logger.info("Flush called. name=" + auditProviderName);
+ super.flush();
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== JSONWriter.flush()");
+ }
+ }
+
+ @Override
+ public void start() {
+ // nothing to start
+ }
+
+ @Override
+ synchronized public void stop() {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> JSONWriter.stop()");
+ }
+ if (logWriter != null) {
+ try {
+ logWriter.flush();
+ logWriter.close();
+ } catch (Throwable t) {
+ logger.error("Error on closing log writter. Exception will be ignored. name="
+ + auditProviderName + ", fileName=" + currentFileName);
+ }
+ logWriter = null;
+ ostream = null;
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== JSONWriter.stop()");
+ }
+ }
+}
\ No newline at end of file
diff --git a/agents-audit/src/main/java/org/apache/ranger/audit/utils/RangerORCAuditWriter.java b/agents-audit/src/main/java/org/apache/ranger/audit/utils/RangerORCAuditWriter.java
new file mode 100644
index 0000000..a3cd140
--- /dev/null
+++ b/agents-audit/src/main/java/org/apache/ranger/audit/utils/RangerORCAuditWriter.java
@@ -0,0 +1,189 @@
+/*
+ * 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.ranger.audit.utils;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.orc.Writer;
+import org.apache.ranger.audit.model.AuthzAuditEvent;
+import org.apache.ranger.audit.provider.MiscUtil;
+
+import java.io.File;
+import java.security.PrivilegedExceptionAction;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * This class writes the Ranger audits to HDFS as ORC files
+ * Refer README.TXT for enabling ORCWriter.
+ */
+public class RangerORCAuditWriter extends AbstractRangerAuditWriter {
+ private static final Log logger = LogFactory.getLog(RangerORCAuditWriter.class);
+
+ protected static final String ORC_FILE_EXTENSION = ".orc";
+ protected volatile ORCFileUtil orcFileUtil = null;
+ protected Writer orcLogWriter = null;
+ protected String fileType = "orc";
+ protected String compression = null;
+ protected int orcBufferSize = 0;
+ protected int defaultbufferSize = 100000;
+ protected long orcStripeSize = 0;
+ protected long defaultStripeSize = 100000L;
+
+ @Override
+ public void init(Properties props, String propPrefix, String auditProviderName, Map<String,String> auditConfigs) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerORCAuditWriter.init()");
+ }
+ init(props,propPrefix);
+ super.init(props, propPrefix, auditProviderName, auditConfigs);
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerORCAuditWriter.init()");
+ }
+ }
+
+ synchronized public boolean logAuditAsORC(final Collection<AuthzAuditEvent> events) throws Exception {
+ boolean ret = false;
+ Writer out = null;
+ try {
+ if (logger.isDebugEnabled()) {
+ logger.debug("UGI=" + MiscUtil.getUGILoginUser()
+ + ". Will write to HDFS file=" + currentFileName);
+ }
+
+ out = MiscUtil.executePrivilegedAction(new PrivilegedExceptionAction<Writer>() {
+ @Override
+ public Writer run() throws Exception {
+ Writer out = getORCFileWrite();
+ orcFileUtil.log(out,events);
+ return out;
+ };
+ });
+ } catch (Exception e) {
+ orcLogWriter = null;
+ logger.error("Error while writing into ORC FileWriter", e);
+ throw e;
+ } finally {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Flushing HDFS audit in ORC Format. Event Size:" + events.size());
+ }
+ if (out != null) {
+ try {
+ //flush and close the ORC batch file
+ orcFileUtil.close(out);
+ ret = true;
+ } catch (Exception e) {
+ logger.error("Error while closing the ORC FileWriter", e);
+ throw e;
+ }
+ orcLogWriter = null;
+ }
+ }
+ return ret;
+ }
+
+ @Override
+ public void flush() {
+ //For HDFSAuditDestionation with ORC format each file is flushed immediately after writing the ORC batch.
+ //So nothing to flush.
+ }
+
+ @Override
+ public boolean log(Collection<String> events) throws Exception {
+ return logAsORC(events);
+ }
+
+ @Override
+ public void start() {
+ // Nothing to do here. We will open the file when the first log request comes
+ }
+
+ @Override
+ synchronized public void stop() {
+ if (orcLogWriter != null) {
+ try {
+ orcFileUtil.close(orcLogWriter);
+ } catch (Throwable t) {
+ logger.error("Error on closing log ORC Writer. Exception will be ignored. name="
+ + auditProviderName + ", fileName=" + currentFileName);
+ }
+ orcLogWriter = null;
+ }
+ }
+
+ @Override
+ public boolean logFile(File file) throws Exception {
+ return false;
+ }
+
+ // Creates ORC Write file
+ synchronized protected Writer getORCFileWrite() throws Exception {
+ if (logger.isDebugEnabled()) {
+ logger.debug("==> RangerORCAuditWriter.getORCFileWrite()");
+ }
+ if (orcLogWriter == null) {
+ // Create the file to write
+ createFileSystemFolders();
+ logger.info("Creating new log file. hdfPath=" + fullPath);
+ orcLogWriter = orcFileUtil.createWriter(conf, fileSystem, fullPath);
+ currentFileName = fullPath;
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("<== RangerORCAuditWriter.getORCFileWrite()");
+ }
+ return orcLogWriter;
+ }
+
+ public boolean logAsORC(Collection<String> events) throws Exception {
+ boolean ret = false;
+ Collection<AuthzAuditEvent> authzAuditEvents = getAuthzAuditEvents(events);
+ ret = logAuditAsORC(authzAuditEvents);
+ return ret;
+ }
+
+ public Collection<AuthzAuditEvent> getAuthzAuditEvents(Collection<String> events) throws Exception {
+ Collection<AuthzAuditEvent> ret = new ArrayList<>();
+ for (String event : events) {
+ try {
+ AuthzAuditEvent authzAuditEvent = MiscUtil.fromJson(event, AuthzAuditEvent.class);
+ ret.add(authzAuditEvent);
+ } catch (Exception e) {
+ logger.error("Error converting to From JSON to AuthzAuditEvent=" + event);
+ throw e;
+ }
+ }
+ return ret;
+ }
+
+ public void init(Properties props, String propPrefix) {
+ compression = MiscUtil.getStringProperty(props, propPrefix + "." + fileType +".compression");
+ orcBufferSize = MiscUtil.getIntProperty(props, propPrefix + "." + fileType +".buffersize",defaultbufferSize);
+ orcStripeSize = MiscUtil.getLongProperty(props, propPrefix + "." + fileType +".stripesize",defaultStripeSize);
+ setFileExtension(ORC_FILE_EXTENSION);
+ try {
+ orcFileUtil = ORCFileUtil.getInstance();
+ orcFileUtil.init(orcBufferSize, orcStripeSize, compression);
+ } catch ( Exception e) {
+ logger.error("Error while doing ORCWriter.init() ", e);
+ }
+ }
+}
\ No newline at end of file