Versions Compared

Old Version 2

changes.mady.by.user Edward Zhang

Saved on

compared with

New Version Current

changes.mady.by.user Edward Zhang

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.
Comment: Migrated to Confluence 5.3

...

Enabling audits logging is as simple as adding some configurations in logging properties files and adding a few kafka libraries.

which the underlying hadoop product supports. Here is an example of configuring Hive logging properties file to enable sending audit events to Kafka messaging bus. 

Code Block
titlehive-log4j.properties
#
# kafka Appender
#
### for Ranger 0.4.0
log4j.logger.com.xasecure.audit.provider.Log4jAuditProvider=INFO,KAFKA_HIVE_AUDIT
log4j.appender.KAFKA_HIVE_AUDIT=kafka.producer.KafkaLog4jAppender
log4j.appender.KAFKA_HIVE_AUDIT.BrokerList=sandbox.hortonworks.com:6667
log4j.appender.KAFKA_HIVE_AUDIT.Topic=hive_audit_log
log4j.appender.KAFKA_HIVE_AUDIT.layout=org.apache.log4j.PatternLayout
log4j.appender.KAFKA_HIVE_AUDIT.layout.ConversionPattern=%d{ISO8601} %-5p [%t]: %c{2} (%F:%M(%L)) - %m%n
log4j.appender.KAFKA_HIVE_AUDIT.ProducerType=async

For Kafka 0.8.1.x, the following jar files should be put into $HIVE_DIR/lib

Code Block
titlekafka libraries
kafka_2.10-0.8.1.2.2.4.2-2.jar
scala-library-2.10.4.jar
metrics-core-2.2.0.jar

 

In consumer side, some deserializer should be there to parse normalized audit records

Code Block
languagejava
titlehive audit parser
public class AuditLogJsonDeserializer extends JsonDeserializer<HiveAuditLogDataModel> {
	private static Logger LOG = LoggerFactory.getLogger(AuditLogJsonDeserializer.class);
	@Override
	public HiveAuditLogDataModel deserialize(JsonParser jp, DeserializationContext ctxt) 
			throws IOException, JsonProcessingException {
		HiveAuditLogDataModel model = new HiveAuditLogDataModel();
		JsonNode node = jp.getCodec().readTree(jp);
		String resource = node.get("resource").asText();
		// split resource to database, table, and column
		String[] tmp = resource.split("/");
		if(tmp.length >= 1){
			model.db = tmp[0];
			if(tmp.length >= 3){
				model.table = tmp[1];
				model.column = tmp[2];
			}
		}
		model.action = node.get("action").asText();
		model.clientIP = node.get("cliIP").asText();
		SimpleDateFormat formatter = new SimpleDateFormat("yyyyMMdd-HH:mm:ss.SSS-Z");
		try{
			model.timestamp = formatter.parse(node.get("evtTime").asText()).getTime();
		}catch(Exception ex){
			LOG.error("fail converting evtTime in hive audit log", ex);
		}
		model.user = node.get("reqUser").asText();
		return model;
	}
}