Asynchronous logging using Log4j, ActiveMQ and Spring

My team and I are creating a services platform based on a set of RESTful JSON services where each service contributes to the platform by providing distinct feature(s) and/or data. With logs being generated all over the place, we thought it was a good idea to centralize logging and perhaps also provide a rudimentary log viewer that allowed us to view, filter, sort and search our logs. We also wanted our logging to be asynchronous as we didn’t want our services to be held up while trying to write logs say maybe directly to a database.

The strategy for achieving this was straight forward.

1. Setup ActiveMQ
2. Create a log4j appender that writes logs to the queue (log4j ships with one such appender but lets write our own.
3. Write a message listener that reads logs from a JMS queue setup on an MQ server and persists them

Let’s take a look one by one.

Setup ActiveMQ

Setting up an external ActiveMQ server is simple enough. A great tutorial is available at http://servicebus.blogspot.com/2011/02/installing-apache-active-mq-on-ubuntu.html to set it up on Ubuntu. You can also choose to embed a message broker within your application.  Spring makes this easy. We will see how later.

Creating a Lo4j JMS appender

First, we create a log4j JMS appender. log4j ships with one such appender (that writes to a JMS topic instead of a queue)

import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.log4j.Appender;
import org.apache.log4j.AppenderSkeleton;
import org.apache.log4j.Logger;
import org.apache.log4j.PatternLayout;
import org.apache.log4j.spi.LoggingEvent;

/**
 * JMSQueue appender is a log4j appender that writes LoggingEvent to a queue.
 * @author faheem
 *
 */
public class JMSQueueAppender extends AppenderSkeleton implements Appender{

private static Logger logger = Logger.getLogger("JMSQueueAppender");

private String brokerUri;
private String queueName;

@Override
public void close() {

}

@Override
public boolean requiresLayout() {
	return false;
}

@Override
protected synchronized void append(LoggingEvent event) {

   try {

     ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
					this.brokerUri);

     // Create a Connection
     javax.jms.Connection connection = connectionFactory.createConnection();
     connection.start();np

     // Create a Session
     Session session = connection.createSession(false,Session.AUTO_ACKNOWLEDGE);

     // Create the destination (Topic or Queue)
     Destination destination = session.createQueue(this.queueName);

     // Create a MessageProducer from the Session to the Topic or Queue
     MessageProducer producer = session.createProducer(destination);
     producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

     ObjectMessage message = session.createObjectMessage(new LoggingEventWrapper(event));

     // Tell the producer to send the message
     producer.send(message);

     // Clean up
     session.close();
     connection.close();
  } catch (Exception e) {
     e.printStackTrace();
  }
}

public void setBrokerUri(String brokerUri) {
	this.brokerUri = brokerUri;
}

public String getBrokerUri() {
	return brokerUri;
}

public void setQueueName(String queueName) {
	this.queueName = queueName;
}

public String getQueueName() {
	return queueName;
}
}

Lets see whats happening here.

Line 19: We implement the Log4J appender interface that asks us to implement three methods. requiresLayout, close and append. We will keep things simple for the moment and implement the append method which gets called whenever a method call to the logger is made.

Line 37: log4j calls the append method and passes a LoggingEvent object as a parameter which represents a call to a logger. A LoggingEvent object encapsulates all information about every log item.

Line 41 & 42: Create a new connection factory by providing it with a uri of a JMS, in our case activemq, server

Line 45, 46 and 49: We establish a connection and a session to the JMS server.  A Session can be opened in several modes.  An Auto_Acknowledge session is one in which the acknowledgment of message happens automatically. Other modes include Client_Acknowledge in which a client has to explicitly acknowledge receipt and/or processing of a message and two other modes. For details, refer to the docs at http://download.oracle.com/javaee/1.4/api/javax/jms/Session.html

Line 52: Create a queue. Send the queue name to connect to as a parameter.

Line 56: We set the delivery mode to Non_Persistent. The other option is Persistent where the message is persisted to a persistent store. Persistent mode slows down but adds reliability to the message transfer.

Line 58: We are doing multiple things. First of all I am wrapping the LoggingEvent object into a LoggingEventWrapper. This is because there are some properties within the LoggingEvent object that are not serializeable and also because I want to capture some additional information such as IP address and host name.  Next, using the JMS session object, I prepare an object (the wrapper) for transport.

Line 61: I send the object to the queue.

Below is the code for the wrapper.

import java.io.Serializable;
import java.net.InetAddress;
import java.net.UnknownHostException;

import org.apache.log4j.EnhancedPatternLayout;
import org.apache.log4j.spi.LoggingEvent;

/**
 * Logging Event Wraps a log4j LoggingEvent object. Wrapping is required by some information is lost
 * when the LoggingEvent is serialized. The idea is to extract all information required from the LoggingEvent
 * object, place it in the wrapper and then serialize the LoggingEventWrapper. This way all required data remains
 * available to us.
 * @author faheem
 *
 */

public class LoggingEventWrapper implements Serializable{

	private static final String ENHANCED_PATTERN_LAYOUT = "%throwable";
	private static final long serialVersionUID = 3281981073249085474L;
	private LoggingEvent loggingEvent;

	private Long timeStamp;
	private String level;
	private String logger;
	private String message;
	private String detail;
	private String ipAddress;
	private String hostName;

	public LoggingEventWrapper(LoggingEvent loggingEvent){
		this.loggingEvent = loggingEvent;

		//Format event and set detail field
	    EnhancedPatternLayout layout = new EnhancedPatternLayout();
	    layout.setConversionPattern(ENHANCED_PATTERN_LAYOUT);
	    this.detail = layout.format(this.loggingEvent);
	}

	public Long getTimeStamp() {
		return this.loggingEvent.timeStamp;
	}

	public String getLevel() {
		return this.loggingEvent.getLevel().toString();
	}

	public String getLogger() {
		return this.loggingEvent.getLoggerName();
	}

	public String getMessage() {
		return this.loggingEvent.getRenderedMessage();
	}

	public String getDetail() {
	    return this.detail;
	}

	public LoggingEvent getLoggingEvent() {
		return loggingEvent;
	}

	public String getIpAddress() {
		try {
			return InetAddress.getLocalHost().getHostAddress();
		} catch (UnknownHostException e) {
			return "Could not determine IP";
		}
	}

	public String getHostName() {
		try {
			return InetAddress.getLocalHost().getHostName();
		} catch (UnknownHostException e) {
			return "Could not determine Host Name";
		}
	}
}

The Message Listener

The message listener “listens” to the queue (or topic). Whenever a new message is added to the queue, the onMessage method is called.

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;

import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class LogQueueListener implements MessageListener
{
	public static Logger logger = Logger.getLogger(LogQueueListener.class);

	@Autowired
	private ILoggingService loggingService;

    public void onMessage( final Message message )
    {
        if ( message instanceof ObjectMessage )
        {
            try{
        	    final LoggingEventWrapper loggingEventWrapper = (LoggingEventWrapper)((ObjectMessage) message).getObject();
        	    loggingService.saveLog(loggingEventWrapper);
            }
            catch (final JMSException e)
            {
                logger.error(e.getMessage(), e);
            } catch (Exception e) {
	        logger.error(e.getMessage(),e);
	    }
        }
    }
}

Line 23: Checking if the object being picked off the queue is an instance of ObjectMessage
Line 26: Extracting LoggingEventWrapper from the Message
Line 27: Call a service method to persist the log

Wiring up in Spring

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:p="http://www.springframework.org/schema/p" xmlns:context="http://www.springframework.org/schema/context" xmlns:jms="http://www.springframework.org/schema/jms" xmlns:amq="http://activemq.apache.org/schema/core" xmlns:aop="http://www.springframework.org/schema/aop" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd http://www.springframework.org/schema/jms http://www.springframework.org/schema/jms/spring-jms-3.0.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core-5.5.0.xsd">
<!-- lets create an embedded ActiveMQ Broker -->
<!-- uncomment the tag below only if you need to create an embedded broker -->
<!-- amq:broker useJmx="false" persistent="false">
     <amq:transportConnectors>
     <amq:transportConnector uri="tcp://localhost:61616" />
     </amq:transportConnectors>
</amq:broker-->
<!-- ActiveMQ destinations to use -->
<amq:queue id="destination" physicalName="logQueue" />
<!-- JMS ConnectionFactory to use, configuring the embedded broker using XML -->
<amq:connectionFactory id="jmsFactory" brokerURL="tcp://localhost:61616" />
<bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory">
   <constructor-arg ref="jmsFactory" />
   <property name="exceptionListener" ref="JMSExceptionListener" />
   <property name="sessionCacheSize" value="100" />
</bean>
<!-- Spring JMS Template -->
<bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
   <constructor-arg ref="connectionFactory" />
</bean>
<!-- listener container definition using the jms namespace, concurrency
is the max number of concurrent listeners that can be started -->
<jms:listener-container concurrency="10">
   <jms:listener id="QueueListener" destination="logQueue" ref="logQueueListener" />
</jms:listener-container>
</beans>

Lines 5-9: Use the broker tag to setup an embedded message broker. Since I am using an external one, I don’t need it.
Line 12: Mention the name of the queue you want to connect to.
Line 14: URI of the Broker Server.
Line 15-19: Connection Factory setup
Line 26-28: Message Listener Setup where we specify the number of concurrent threads that can consume messages off the queue.

Of course, the above example will not work out of the box. You still have to include all JMS dependencies and implement the service that persists logs. But I hope it gives you a decent idea.