src/ListenSyslog.cpp - nifi-minifi-cpp - Git at Google

 /**
  * @file ListenSyslog.cpp
  * ListenSyslog class implementation
  *
  * 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.
  */
 #include <queue>
 #include <stdio.h>
 #include <string>
 #include "TimeUtil.h"
 #include "ListenSyslog.h"
 #include "ProcessContext.h"
 #include "ProcessSession.h"

 const std::string ListenSyslog::ProcessorName("ListenSyslog");
 Property ListenSyslog::RecvBufSize("Receive Buffer Size", "The size of each buffer used to receive Syslog messages.", "65507 B");
 Property ListenSyslog::MaxSocketBufSize("Max Size of Socket Buffer", "The maximum size of the socket buffer that should be used.", "1 MB");
 Property ListenSyslog::MaxConnections("Max Number of TCP Connections", "The maximum number of concurrent connections to accept Syslog messages in TCP mode.", "2");
 Property ListenSyslog::MaxBatchSize("Max Batch Size",
 		"The maximum number of Syslog events to add to a single FlowFile.", "1");
 Property ListenSyslog::MessageDelimiter("Message Delimiter",
 		"Specifies the delimiter to place between Syslog messages when multiple messages are bundled together (see <Max Batch Size> property).", "\n");
 Property ListenSyslog::ParseMessages("Parse Messages",
 		"Indicates if the processor should parse the Syslog messages. If set to false, each outgoing FlowFile will only.", "false");
 Property ListenSyslog::Protocol("Protocol", "The protocol for Syslog communication.", "UDP");
 Property ListenSyslog::Port("Port", "The port for Syslog communication.", "514");
 Relationship ListenSyslog::Success("success", "All files are routed to success");
 Relationship ListenSyslog::Invalid("invalid", "SysLog message format invalid");

 void ListenSyslog::initialize()
 {
 	//! Set the supported properties
 	std::set<Property> properties;
 	properties.insert(RecvBufSize);
 	properties.insert(MaxSocketBufSize);
 	properties.insert(MaxConnections);
 	properties.insert(MaxBatchSize);
 	properties.insert(MessageDelimiter);
 	properties.insert(ParseMessages);
 	properties.insert(Protocol);
 	properties.insert(Port);
 	setSupportedProperties(properties);
 	//! Set the supported relationships
 	std::set<Relationship> relationships;
 	relationships.insert(Success);
 	relationships.insert(Invalid);
 	setSupportedRelationships(relationships);
 }

 void ListenSyslog::startSocketThread()
 {
 	if (_thread != NULL)
 		return;

 	_logger->log_info("ListenSysLog Socket Thread Start");
 	_serverTheadRunning = true;
 	_thread = new std::thread(run, this);
 	_thread->detach();
 }

 void ListenSyslog::run(ListenSyslog *process)
 {
 	process->runThread();
 }

 void ListenSyslog::runThread()
 {
 	while (_serverTheadRunning)
 	{
 		if (_resetServerSocket)
 		{
 			_resetServerSocket = false;
 			// need to reset the socket
 			std::vector<int>::iterator it;
 			for (it = _clientSockets.begin(); it != _clientSockets.end(); ++it)
 			{
 				int clientSocket = *it;
 				close(clientSocket);
 			}
 			_clientSockets.clear();
 			if (_serverSocket > 0)
 			{
 				close(_serverSocket);
 				_serverSocket = 0;
 			}
 		}

 		if (_serverSocket <= 0)
 		{
 			uint16_t portno = _port;
 			struct sockaddr_in serv_addr;
 			int sockfd;
 			if (_protocol == "TCP")
 				sockfd = socket(AF_INET, SOCK_STREAM, 0);
 			else
 				sockfd = socket(AF_INET, SOCK_DGRAM, 0);
 			if (sockfd < 0)
 			{
 				_logger->log_info("ListenSysLog Server socket creation failed");
 				break;
 			}
 			bzero((char *) &serv_addr, sizeof(serv_addr));
 			serv_addr.sin_family = AF_INET;
 			serv_addr.sin_addr.s_addr = INADDR_ANY;
 			serv_addr.sin_port = htons(portno);
 			if (bind(sockfd, (struct sockaddr *) &serv_addr,
 					sizeof(serv_addr)) < 0)
 			{
 				_logger->log_error("ListenSysLog Server socket bind failed");
 				break;
 			}
 			if (_protocol == "TCP")
 				listen(sockfd,5);
 			_serverSocket = sockfd;
 			_logger->log_error("ListenSysLog Server socket %d bind OK to port %d", _serverSocket, portno);
 		}
 		FD_ZERO(&_readfds);
 		FD_SET(_serverSocket, &_readfds);
 		_maxFds = _serverSocket;
 		std::vector<int>::iterator it;
 		for (it = _clientSockets.begin(); it != _clientSockets.end(); ++it)
 		{
 			int clientSocket = *it;
 			if (clientSocket >= _maxFds)
 				_maxFds = clientSocket;
 			FD_SET(clientSocket, &_readfds);
 		}
 		fd_set fds;
 		struct timeval tv;
 		int retval;
 		fds = _readfds;
 		tv.tv_sec = 0;
 		// 100 msec
 		tv.tv_usec = 100000;
 		retval = select(_maxFds+1, &fds, NULL, NULL, &tv);
 		if (retval < 0)
 			break;
 		if (retval == 0)
 			continue;
 		if (FD_ISSET(_serverSocket, &fds))
 		{
 			// server socket, either we have UDP datagram or TCP connection request
 			if (_protocol == "TCP")
 			{
 				socklen_t clilen;
 				struct sockaddr_in cli_addr;
 				clilen = sizeof(cli_addr);
 				int newsockfd = accept(_serverSocket,
 						(struct sockaddr *) &cli_addr,
 						&clilen);
 				if (newsockfd > 0)
 				{
 					if (_clientSockets.size() < _maxConnections)
 					{
 						_clientSockets.push_back(newsockfd);
 						_logger->log_info("ListenSysLog new client socket %d connection", newsockfd);
 						continue;
 					}
 					else
 					{
 						close(newsockfd);
 					}
 				}
 			}
 			else
 			{
 				socklen_t clilen;
 				struct sockaddr_in cli_addr;
 				clilen = sizeof(cli_addr);
 				int recvlen = recvfrom(_serverSocket, _buffer, sizeof(_buffer), 0,
 						(struct sockaddr *)&cli_addr, &clilen);
 				if (recvlen > 0 && (recvlen + getEventQueueByteSize()) <= _recvBufSize)
 				{
 					uint8_t *payload = new uint8_t[recvlen];
 					memcpy(payload, _buffer, recvlen);
 					putEvent(payload, recvlen);
 				}
 			}
 		}
 		it = _clientSockets.begin();
 		while (it != _clientSockets.end())
 		{
 			int clientSocket = *it;
 			if (FD_ISSET(clientSocket, &fds))
 			{
 				int recvlen = readline(clientSocket, (char *)_buffer, sizeof(_buffer));
 				if (recvlen <= 0)
 				{
 					close(clientSocket);
 					_logger->log_info("ListenSysLog client socket %d close", clientSocket);
 					it = _clientSockets.erase(it);
 				}
 				else
 				{
 					if ((recvlen + getEventQueueByteSize()) <= _recvBufSize)
 					{
 						uint8_t *payload = new uint8_t[recvlen];
 						memcpy(payload, _buffer, recvlen);
 						putEvent(payload, recvlen);
 					}
 					++it;
 				}
 			}
 		}
 	}
 	return;
 }


 int ListenSyslog::readline( int fd, char *bufptr, size_t len )
 {
 	char *bufx = bufptr;
 	static char *bp;
 	static int cnt = 0;
 	static char b[ 2048 ];
 	char c;

 	while ( --len > 0 )
     {
       if ( --cnt <= 0 )
       {
     	  cnt = recv( fd, b, sizeof( b ), 0 );
     	  if ( cnt < 0 )
     	  {
     		  if ( errno == EINTR )
     		  {
     			  len++;		/* the while will decrement */
     			  continue;
     		  }
     		  return -1;
     	  }
     	  if ( cnt == 0 )
     		  return 0;
     	  bp = b;
       }
       c = *bp++;
       *bufptr++ = c;
       if ( c == '\n' )
       {
     	  *bufptr = '\n';
     	  return bufptr - bufx + 1;
       }
     }
 	return -1;
 }

 void ListenSyslog::onTrigger(ProcessContext *context, ProcessSession *session)
 {
 	std::string value;
 	bool needResetServerSocket = false;
 	if (context->getProperty(Protocol.getName(), value))
 	{
 		if (_protocol != value)
 			needResetServerSocket = true;
 		_protocol = value;
 	}
 	if (context->getProperty(RecvBufSize.getName(), value))
 	{
 		Property::StringToInt(value, _recvBufSize);
 	}
 	if (context->getProperty(MaxSocketBufSize.getName(), value))
 	{
 		Property::StringToInt(value, _maxSocketBufSize);
 	}
 	if (context->getProperty(MaxConnections.getName(), value))
 	{
 		Property::StringToInt(value, _maxConnections);
 	}
 	if (context->getProperty(MessageDelimiter.getName(), value))
 	{
 		_messageDelimiter = value;
 	}
 	if (context->getProperty(ParseMessages.getName(), value))
 	{
 		Property::StringToBool(value, _parseMessages);
 	}
 	if (context->getProperty(Port.getName(), value))
 	{
 		int64_t oldPort = _port;
 		Property::StringToInt(value, _port);
 		if (_port != oldPort)
 			needResetServerSocket = true;
 	}
 	if (context->getProperty(MaxBatchSize.getName(), value))
 	{
 		Property::StringToInt(value, _maxBatchSize);
 	}

 	if (needResetServerSocket)
 		_resetServerSocket = true;

 	startSocketThread();

 	// read from the event queue
 	if (isEventQueueEmpty())
 	{
 		context->yield();
 		return;
 	}

 	std::queue<SysLogEvent> eventQueue;
 	pollEvent(eventQueue, _maxBatchSize);
 	bool firstEvent = true;
 	FlowFileRecord *flowFile = NULL;
 	while(!eventQueue.empty())
 	{
 		SysLogEvent event = eventQueue.front();
 		eventQueue.pop();
 		if (firstEvent)
 		{
 			flowFile = session->create();
 			if (!flowFile)
 				return;
 			ListenSyslog::WriteCallback callback((char *)event.payload, event.len);
 			session->write(flowFile, &callback);
 			delete[] event.payload;
 			firstEvent = false;
 		}
 		else
 		{
 			ListenSyslog::WriteCallback callback((char *)event.payload, event.len);
 			session->append(flowFile, &callback);
 			delete[] event.payload;
 		}
 	}
 	flowFile->addAttribute("syslog.protocol", _protocol);
 	flowFile->addAttribute("syslog.port", std::to_string(_port));
 	session->transfer(flowFile, Success);
 }
	/**
	* @file ListenSyslog.cpp
	* ListenSyslog class implementation
	*
	* 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.
	*/
	#include <queue>
	#include <stdio.h>
	#include <string>
	#include "TimeUtil.h"
	#include "ListenSyslog.h"
	#include "ProcessContext.h"
	#include "ProcessSession.h"

	const std::string ListenSyslog::ProcessorName("ListenSyslog");
	Property ListenSyslog::RecvBufSize("Receive Buffer Size", "The size of each buffer used to receive Syslog messages.", "65507 B");
	Property ListenSyslog::MaxSocketBufSize("Max Size of Socket Buffer", "The maximum size of the socket buffer that should be used.", "1 MB");
	Property ListenSyslog::MaxConnections("Max Number of TCP Connections", "The maximum number of concurrent connections to accept Syslog messages in TCP mode.", "2");
	Property ListenSyslog::MaxBatchSize("Max Batch Size",
	"The maximum number of Syslog events to add to a single FlowFile.", "1");
	Property ListenSyslog::MessageDelimiter("Message Delimiter",
	"Specifies the delimiter to place between Syslog messages when multiple messages are bundled together (see <Max Batch Size> property).", "\n");
	Property ListenSyslog::ParseMessages("Parse Messages",
	"Indicates if the processor should parse the Syslog messages. If set to false, each outgoing FlowFile will only.", "false");
	Property ListenSyslog::Protocol("Protocol", "The protocol for Syslog communication.", "UDP");
	Property ListenSyslog::Port("Port", "The port for Syslog communication.", "514");
	Relationship ListenSyslog::Success("success", "All files are routed to success");
	Relationship ListenSyslog::Invalid("invalid", "SysLog message format invalid");

	void ListenSyslog::initialize()
	{
	//! Set the supported properties
	std::set<Property> properties;
	properties.insert(RecvBufSize);
	properties.insert(MaxSocketBufSize);
	properties.insert(MaxConnections);
	properties.insert(MaxBatchSize);
	properties.insert(MessageDelimiter);
	properties.insert(ParseMessages);
	properties.insert(Protocol);
	properties.insert(Port);
	setSupportedProperties(properties);
	//! Set the supported relationships
	std::set<Relationship> relationships;
	relationships.insert(Success);
	relationships.insert(Invalid);
	setSupportedRelationships(relationships);
	}

	void ListenSyslog::startSocketThread()
	{
	if (_thread != NULL)
	return;

	_logger->log_info("ListenSysLog Socket Thread Start");
	_serverTheadRunning = true;
	_thread = new std::thread(run, this);
	_thread->detach();
	}

	void ListenSyslog::run(ListenSyslog *process)
	{
	process->runThread();
	}

	void ListenSyslog::runThread()
	{
	while (_serverTheadRunning)
	{
	if (_resetServerSocket)
	{
	_resetServerSocket = false;
	// need to reset the socket
	std::vector<int>::iterator it;
	for (it = _clientSockets.begin(); it != _clientSockets.end(); ++it)
	{
	int clientSocket = *it;
	close(clientSocket);
	}
	_clientSockets.clear();
	if (_serverSocket > 0)
	{
	close(_serverSocket);
	_serverSocket = 0;
	}
	}

	if (_serverSocket <= 0)
	{
	uint16_t portno = _port;
	struct sockaddr_in serv_addr;
	int sockfd;
	if (_protocol == "TCP")
	sockfd = socket(AF_INET, SOCK_STREAM, 0);
	else
	sockfd = socket(AF_INET, SOCK_DGRAM, 0);
	if (sockfd < 0)
	{
	_logger->log_info("ListenSysLog Server socket creation failed");
	break;
	}
	bzero((char *) &serv_addr, sizeof(serv_addr));
	serv_addr.sin_family = AF_INET;
	serv_addr.sin_addr.s_addr = INADDR_ANY;
	serv_addr.sin_port = htons(portno);
	if (bind(sockfd, (struct sockaddr *) &serv_addr,
	sizeof(serv_addr)) < 0)
	{
	_logger->log_error("ListenSysLog Server socket bind failed");
	break;
	}
	if (_protocol == "TCP")
	listen(sockfd,5);
	_serverSocket = sockfd;
	_logger->log_error("ListenSysLog Server socket %d bind OK to port %d", _serverSocket, portno);
	}
	FD_ZERO(&_readfds);
	FD_SET(_serverSocket, &_readfds);
	_maxFds = _serverSocket;
	std::vector<int>::iterator it;
	for (it = _clientSockets.begin(); it != _clientSockets.end(); ++it)
	{
	int clientSocket = *it;
	if (clientSocket >= _maxFds)
	_maxFds = clientSocket;
	FD_SET(clientSocket, &_readfds);
	}
	fd_set fds;
	struct timeval tv;
	int retval;
	fds = _readfds;
	tv.tv_sec = 0;
	// 100 msec
	tv.tv_usec = 100000;
	retval = select(_maxFds+1, &fds, NULL, NULL, &tv);
	if (retval < 0)
	break;
	if (retval == 0)
	continue;
	if (FD_ISSET(_serverSocket, &fds))
	{
	// server socket, either we have UDP datagram or TCP connection request
	if (_protocol == "TCP")
	{
	socklen_t clilen;
	struct sockaddr_in cli_addr;
	clilen = sizeof(cli_addr);
	int newsockfd = accept(_serverSocket,
	(struct sockaddr *) &cli_addr,
	&clilen);
	if (newsockfd > 0)
	{
	if (_clientSockets.size() < _maxConnections)
	{
	_clientSockets.push_back(newsockfd);
	_logger->log_info("ListenSysLog new client socket %d connection", newsockfd);
	continue;
	}
	else
	{
	close(newsockfd);
	}
	}
	}
	else
	{
	socklen_t clilen;
	struct sockaddr_in cli_addr;
	clilen = sizeof(cli_addr);
	int recvlen = recvfrom(_serverSocket, _buffer, sizeof(_buffer), 0,
	(struct sockaddr *)&cli_addr, &clilen);
	if (recvlen > 0 && (recvlen + getEventQueueByteSize()) <= _recvBufSize)
	{
	uint8_t *payload = new uint8_t[recvlen];
	memcpy(payload, _buffer, recvlen);
	putEvent(payload, recvlen);
	}
	}
	}
	it = _clientSockets.begin();
	while (it != _clientSockets.end())
	{
	int clientSocket = *it;
	if (FD_ISSET(clientSocket, &fds))
	{
	int recvlen = readline(clientSocket, (char *)_buffer, sizeof(_buffer));
	if (recvlen <= 0)
	{
	close(clientSocket);
	_logger->log_info("ListenSysLog client socket %d close", clientSocket);
	it = _clientSockets.erase(it);
	}
	else
	{
	if ((recvlen + getEventQueueByteSize()) <= _recvBufSize)
	{
	uint8_t *payload = new uint8_t[recvlen];
	memcpy(payload, _buffer, recvlen);
	putEvent(payload, recvlen);
	}
	++it;
	}
	}
	}
	}
	return;
	}


	int ListenSyslog::readline( int fd, char *bufptr, size_t len )
	{
	char *bufx = bufptr;
	static char *bp;
	static int cnt = 0;
	static char b[ 2048 ];
	char c;

	while ( --len > 0 )
	{
	if ( --cnt <= 0 )
	{
	cnt = recv( fd, b, sizeof( b ), 0 );
	if ( cnt < 0 )
	{
	if ( errno == EINTR )
	{
	len++; /* the while will decrement */
	continue;
	}
	return -1;
	}
	if ( cnt == 0 )
	return 0;
	bp = b;
	}
	c = *bp++;
	*bufptr++ = c;
	if ( c == '\n' )
	{
	*bufptr = '\n';
	return bufptr - bufx + 1;
	}
	}
	return -1;
	}

	void ListenSyslog::onTrigger(ProcessContext context, ProcessSession session)
	{
	std::string value;
	bool needResetServerSocket = false;
	if (context->getProperty(Protocol.getName(), value))
	{
	if (_protocol != value)
	needResetServerSocket = true;
	_protocol = value;
	}
	if (context->getProperty(RecvBufSize.getName(), value))
	{
	Property::StringToInt(value, _recvBufSize);
	}
	if (context->getProperty(MaxSocketBufSize.getName(), value))
	{
	Property::StringToInt(value, _maxSocketBufSize);
	}
	if (context->getProperty(MaxConnections.getName(), value))
	{
	Property::StringToInt(value, _maxConnections);
	}
	if (context->getProperty(MessageDelimiter.getName(), value))
	{
	_messageDelimiter = value;
	}
	if (context->getProperty(ParseMessages.getName(), value))
	{
	Property::StringToBool(value, _parseMessages);
	}
	if (context->getProperty(Port.getName(), value))
	{
	int64_t oldPort = _port;
	Property::StringToInt(value, _port);
	if (_port != oldPort)
	needResetServerSocket = true;
	}
	if (context->getProperty(MaxBatchSize.getName(), value))
	{
	Property::StringToInt(value, _maxBatchSize);
	}

	if (needResetServerSocket)
	_resetServerSocket = true;

	startSocketThread();

	// read from the event queue
	if (isEventQueueEmpty())
	{
	context->yield();
	return;
	}

	std::queue<SysLogEvent> eventQueue;
	pollEvent(eventQueue, _maxBatchSize);
	bool firstEvent = true;
	FlowFileRecord *flowFile = NULL;
	while(!eventQueue.empty())
	{
	SysLogEvent event = eventQueue.front();
	eventQueue.pop();
	if (firstEvent)
	{
	flowFile = session->create();
	if (!flowFile)
	return;
	ListenSyslog::WriteCallback callback((char *)event.payload, event.len);
	session->write(flowFile, &callback);
	delete[] event.payload;
	firstEvent = false;
	}
	else
	{
	ListenSyslog::WriteCallback callback((char *)event.payload, event.len);
	session->append(flowFile, &callback);
	delete[] event.payload;
	}
	}
	flowFile->addAttribute("syslog.protocol", _protocol);
	flowFile->addAttribute("syslog.port", std::to_string(_port));
	session->transfer(flowFile, Success);
	}