poco/Net/src/RemoteSyslogListener.cpp

//
// RemoteSyslogListener.cpp
//
// $Id: //poco/Main/Net/src/RemoteSyslogListener.cpp#7 $
//
// Library: Net
// Package: Logging
// Module:  RemoteSyslogListener
//
// Copyright (c) 2007, Applied Informatics Software Engineering GmbH.
// and Contributors.
//
// Permission is hereby granted, free of charge, to any person or organization
// obtaining a copy of the software and accompanying documentation covered by
// this license (the "Software") to use, reproduce, display, distribute,
// execute, and transmit the Software, and to prepare derivative works of the
// Software, and to permit third-parties to whom the Software is furnished to
// do so, all subject to the following:
// 
// The copyright notices in the Software and this entire statement, including
// the above license grant, this restriction and the following disclaimer,
// must be included in all copies of the Software, in whole or in part, and
// all derivative works of the Software, unless such copies or derivative
// works are solely in the form of machine-executable object code generated by
// a source language processor.
// 
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE, TITLE AND NON-INFRINGEMENT. IN NO EVENT
// SHALL THE COPYRIGHT HOLDERS OR ANYONE DISTRIBUTING THE SOFTWARE BE LIABLE
// FOR ANY DAMAGES OR OTHER LIABILITY, WHETHER IN CONTRACT, TORT OR OTHERWISE,
// ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
// DEALINGS IN THE SOFTWARE.
//


#include "Poco/Net/RemoteSyslogListener.h"
#include "Poco/Net/RemoteSyslogChannel.h"
#include "Poco/Net/DatagramSocket.h"
#include "Poco/Runnable.h"
#include "Poco/Notification.h"
#include "Poco/AutoPtr.h"
#include "Poco/NumberParser.h"
#include "Poco/NumberFormatter.h"
#include "Poco/DateTimeParser.h"
#include "Poco/Message.h"
#include "Poco/LoggingFactory.h"
#include "Poco/Buffer.h"
#include <cctype>
#include <cstddef>


namespace Poco {
namespace Net {


//
// MessageNotification
//


class MessageNotification: public Poco::Notification
{
public:
	MessageNotification(const std::string& msg)
	{
		_msg = msg;
	}
	
	~MessageNotification()
	{
	}
	
	const std::string& message() const
	{
		return _msg;
	}
	
private:
	std::string _msg;
};


//
// RemoteUDPListener
//


class RemoteUDPListener: public Poco::Runnable
{
public:
	enum
	{
		WAITTIME_MILLISEC = 1000,
		BUFFER_SIZE = 65536
	};
	
	RemoteUDPListener(Poco::NotificationQueue& queue, Poco::UInt16 port);
	~RemoteUDPListener();

	void run();
	void safeStop();

private:
	Poco::NotificationQueue& _queue;
	DatagramSocket           _socket;
	bool                     _stopped;
};


RemoteUDPListener::RemoteUDPListener(Poco::NotificationQueue& queue, Poco::UInt16 port):
	_queue(queue),
	_socket(Poco::Net::SocketAddress(Poco::Net::IPAddress(), port)),
	_stopped(true)
{
}


RemoteUDPListener::~RemoteUDPListener()
{
}


void RemoteUDPListener::run()
{
	poco_assert (_stopped);

	Poco::Buffer<char> buffer(BUFFER_SIZE);
	_stopped = false;
	Poco::Timespan waitTime(WAITTIME_MILLISEC* 1000);
	while (!_stopped)
	{
		try
		{
			if (_socket.poll(waitTime, Socket::SELECT_READ))
			{
				int byteCnt = _socket.receiveBytes(buffer.begin(), BUFFER_SIZE);
				if (byteCnt > 0)
				{
					_queue.enqueueNotification(new MessageNotification(std::string(buffer.begin(), byteCnt)));
				}
			}
		}
		catch (...)
		{
			// lazy exception catching
		}
	}
}


void RemoteUDPListener::safeStop()
{
	_stopped = true;
}


//
// SyslogParser
//


class SyslogParser: public Poco::Runnable
{
public:
	static const std::string NILVALUE;

	enum
	{
		WAITTIME_MILLISEC = 1000
	};

	SyslogParser(Poco::NotificationQueue& queue, RemoteSyslogListener* pListener);
	~SyslogParser();

	void run();
	void safeStop();

	static Poco::Message::Priority convert(RemoteSyslogChannel::Severity severity);

private:
	void parse(const std::string& msg);
	void parsePrio(const std::string& msg, std::size_t& pos, RemoteSyslogChannel::Severity& severity, RemoteSyslogChannel::Facility& fac);
	void parseNew(const std::string& msg, RemoteSyslogChannel::Severity severity, RemoteSyslogChannel::Facility fac, std::size_t& pos);
	void parseBSD(const std::string& msg, RemoteSyslogChannel::Severity severity, RemoteSyslogChannel::Facility fac, std::size_t& pos);

	static std::string parseUntilSpace(const std::string& msg, std::size_t& pos);
		/// Parses until it encounters the next space char, returns the string from pos, excluding space
		/// pos will point past the space char

private:
	Poco::NotificationQueue& _queue;
	bool                     _stopped;
	RemoteSyslogListener*    _pListener;
};


const std::string SyslogParser::NILVALUE("-");


SyslogParser::SyslogParser(Poco::NotificationQueue& queue, RemoteSyslogListener* pListener):
	_queue(queue),
	_stopped(true),
	_pListener(pListener)
{
	poco_check_ptr (_pListener);
}


SyslogParser::~SyslogParser()
{
}


void SyslogParser::run()
{
	poco_assert (_stopped);
	_stopped = false;
	while (!_stopped)
	{
		try
		{
			Poco::AutoPtr<Poco::Notification> pNf(_queue.waitDequeueNotification(WAITTIME_MILLISEC));
			if (pNf)
			{
				Poco::AutoPtr<MessageNotification> pMsgNf = pNf.cast<MessageNotification>();
				parse(pMsgNf->message());
			}
		}
		catch (Poco::Exception&)
		{
			// parsing exception, what should we do?
		}
		catch (...)
		{
		}
	}
}


void SyslogParser::safeStop()
{
	_stopped = true;
}


void SyslogParser::parse(const std::string& msg)
{
	// <int> -> int: lower 3 bits severity, upper bits: facility
	std::size_t pos = 0;
	RemoteSyslogChannel::Severity severity;
	RemoteSyslogChannel::Facility fac;
	parsePrio(msg, pos, severity, fac);

	// the next field decide if we parse an old BSD message or a new syslog message
	// BSD: expects a month value in string form: Jan, Feb...
	// SYSLOG expects a version number: 1
	
	if (std::isdigit(msg[pos]))
	{
		parseNew(msg, severity, fac, pos);
	}
	else
	{
		parseBSD(msg, severity, fac, pos);
	}
	poco_assert (pos == msg.size());
}


void SyslogParser::parsePrio(const std::string& msg, std::size_t& pos, RemoteSyslogChannel::Severity& severity, RemoteSyslogChannel::Facility& fac)
{
	poco_assert (pos < msg.size());
	poco_assert (msg[pos] == '<');
	++pos;
	std::size_t start = pos;
	
	while (pos < msg.size() && std::isdigit(msg[pos]))
		++pos;
	
	poco_assert (msg[pos] == '>');
	poco_assert (pos - start > 0);
	std::string valStr = msg.substr(start, pos - start);
	++pos; // skip the >

	int val = Poco::NumberParser::parse(valStr);
	poco_assert (val >= 0 && val <= (RemoteSyslogChannel::SYSLOG_LOCAL7 + RemoteSyslogChannel::SYSLOG_DEBUG));
	
	Poco::UInt16 pri = static_cast<Poco::UInt16>(val);
	// now get the lowest 3 bits
	severity = static_cast<RemoteSyslogChannel::Severity>(pri & 0x0007u);
	fac = static_cast<RemoteSyslogChannel::Facility>(pri & 0xfff8u);
}


void SyslogParser::parseNew(const std::string& msg, RemoteSyslogChannel::Severity severity, RemoteSyslogChannel::Facility fac, std::size_t& pos)
{
	Poco::Message::Priority prio = convert(severity);
	// rest of the unparsed header is:
	// VERSION SP TIMESTAMP SP HOSTNAME SP APP-NAME SP PROCID SP MSGID
	std::string versionStr(parseUntilSpace(msg, pos));
	std::string timeStr(parseUntilSpace(msg, pos)); // can be the nilvalue!
	std::string hostName(parseUntilSpace(msg, pos));
	std::string appName(parseUntilSpace(msg, pos));
	std::string procId(parseUntilSpace(msg, pos));
	std::string msgId(parseUntilSpace(msg, pos));
	std::string message(msg.substr(pos));
	pos = msg.size();
	Poco::DateTime date;
	int tzd = 0;
	bool hasDate = Poco::DateTimeParser::tryParse(RemoteSyslogChannel::SYSLOG_TIMEFORMAT, timeStr, date, tzd);
	Poco::Message logEntry(msgId, message, prio);
	logEntry["host"] = hostName;
	logEntry["app"] = appName;
	
	if (hasDate)
		logEntry.setTime(date.timestamp());
	int lval(0);
	Poco::NumberParser::tryParse(procId, lval);
	logEntry.setPid(lval);
	_pListener->log(logEntry);
}


void SyslogParser::parseBSD(const std::string& msg, RemoteSyslogChannel::Severity severity, RemoteSyslogChannel::Facility fac, std::size_t& pos)
{
	Poco::Message::Priority prio = convert(severity);
	// rest of the unparsed header is:
	// "%b %f %H:%M:%S" SP hostname|ipaddress
	// detect three spaces
	int spaceCnt = 0;
	std::size_t start = pos;
	while (spaceCnt < 3 && pos < msg.size())
	{
		if (msg[pos] == ' ')
		{
			spaceCnt++;
			if (spaceCnt == 1)
			{
				// size must be 3 chars for month
				if (pos - start != 3)
				{
					// probably a shortened time value, or the hostname
					// assume hostName
					Poco::Message logEntry(msg.substr(start, pos-start), msg.substr(pos+1), prio);
					_pListener->log(logEntry);
					return;
				}
			}
			else if (spaceCnt == 2)
			{
				// a day value!
				if (!(std::isdigit(msg[pos-1]) && (std::isdigit(msg[pos-2]) || std::isspace(msg[pos-2]))))
				{
					// assume the next field is a hostname
					spaceCnt = 3;
				}
			}
			if (pos + 1 < msg.size() && msg[pos+1] == ' ')
			{
				// we have two spaces when the day value is smaller than 10!
				++pos; // skip one
			}
		}
		++pos;
	}
	std::string timeStr(msg.substr(start, pos-start-1));
	int tzd(0);
	Poco::DateTime date;
	int year = date.year(); // year is not included, use the current one
	bool hasDate = Poco::DateTimeParser::tryParse(RemoteSyslogChannel::BSD_TIMEFORMAT, timeStr, date, tzd);
	if (hasDate)
	{
		int m = date.month();
		int d = date.day();
		int h = date.hour();
		int min = date.minute();
		int sec = date.second();
		date = Poco::DateTime(year, m, d, h, min, sec);
	}
	// next entry is host SP
	std::string hostName(parseUntilSpace(msg, pos));

	// TAG: at most 32 alphanumeric chars, ANY non alphannumeric indicates start of message content
	// ignore: treat everything as content
	std::string message(msg.substr(pos));
	pos = msg.size();
	Poco::Message logEntry(hostName, message, prio);
	logEntry.setTime(date.timestamp());
	_pListener->log(logEntry);
}


std::string SyslogParser::parseUntilSpace(const std::string& msg, std::size_t& pos)
{
	std::size_t start = pos;
	while (pos < msg.size() && !std::isspace(msg[pos]))
		++pos;
	// skip space
	++pos;
	return msg.substr(start, pos-start-1);
}


Poco::Message::Priority SyslogParser::convert(RemoteSyslogChannel::Severity severity)
{
	switch (severity)
	{
	case RemoteSyslogChannel::SYSLOG_EMERGENCY:
		return Poco::Message::PRIO_FATAL;
	case RemoteSyslogChannel::SYSLOG_ALERT:
		return Poco::Message::PRIO_FATAL;
	case RemoteSyslogChannel::SYSLOG_CRITICAL:
		return Poco::Message::PRIO_CRITICAL;
	case RemoteSyslogChannel::SYSLOG_ERROR:
		return Poco::Message::PRIO_ERROR;
	case RemoteSyslogChannel::SYSLOG_WARNING:
		return Poco::Message::PRIO_WARNING;
	case RemoteSyslogChannel::SYSLOG_NOTICE:
		return Poco::Message::PRIO_NOTICE;
	case RemoteSyslogChannel::SYSLOG_INFORMATIONAL:
		return Poco::Message::PRIO_INFORMATION;
	case RemoteSyslogChannel::SYSLOG_DEBUG:
		return Poco::Message::PRIO_DEBUG;
	}
	throw Poco::LogicException("Illegal severity value in message");
}


//
// RemoteSyslogListener
//


const std::string RemoteSyslogListener::PROP_PORT("port");


RemoteSyslogListener::RemoteSyslogListener():
	_pListener(0),
	_pParser(0),
	_listener(),
	_parser(),
	_queue(),
	_port(RemoteSyslogChannel::SYSLOG_PORT)
{
}


RemoteSyslogListener::RemoteSyslogListener(Poco::UInt16 port):
	_pListener(0),
	_pParser(0),
	_listener(),
	_parser(),
	_queue(),
	_port(port)
{
}


RemoteSyslogListener::~RemoteSyslogListener()
{
}


void RemoteSyslogListener::setProperty(const std::string& name, const std::string& value)
{
	if (name == PROP_PORT)
	{
		int val = Poco::NumberParser::parse(value);
		if (val > 0 && val < 65536)
			_port = static_cast<Poco::UInt16>(val);
		else
			throw Poco::InvalidArgumentException("Not a valid port number", value);
	}
	else 
	{
		SplitterChannel::setProperty(name, value);
	}
}


std::string RemoteSyslogListener::getProperty(const std::string& name) const
{
	if (name == PROP_PORT)
		return Poco::NumberFormatter::format(_port);
	else	
		return SplitterChannel::getProperty(name);
}


void RemoteSyslogListener::open()
{
	SplitterChannel::open();
	_pParser = new SyslogParser(_queue, this);
	_pListener = new RemoteUDPListener(_queue, _port);
	_parser.start(*_pParser);
	_listener.start(*_pListener);
}


void RemoteSyslogListener::close()
{
	if (_pListener && _pParser)
	{
		_pListener->safeStop();
		_pParser->safeStop();
		_queue.clear();
		_listener.join();
		_parser.join();
		delete _pListener;
		delete _pParser;
		_pListener = 0;
		_pParser = 0;
	}
	SplitterChannel::close();
}


void RemoteSyslogListener::registerChannel()
{
	Poco::LoggingFactory::defaultFactory().registerChannelClass("RemoteSyslogListener", new Poco::Instantiator<RemoteSyslogListener, Poco::Channel>);
}


} } // namespace Poco::Net