dcdbpusher.cpp

//================================================================================
// Name        : dcdbpusher.cpp
// Author      : Michael Ott (original), Micha Mueller
// Copyright   : Leibniz Supercomputing Centre
// Description : Main functions for the DCDB MQTT Pusher
//================================================================================

//================================================================================
// This file is part of DCDB (DataCenter DataBase)
// Copyright (C) 2011-2016 Leibniz Supercomputing Centre
//
// This program is free software; you can redistribute it and/or
// modify it under the terms of the GNU General Public License
// as published by the Free Software Foundation; either version 2
// of the License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program; if not, write to the Free Software
// Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
//================================================================================

#include <dcdbdaemon.h>
#include <functional>
#include <string>

//Caution: include order matters! HttpsServer.h needs to be included first
#include "HttpsServer.h"
#include "Sensor.h"
#include "Configuration.h"
#include "MQTTPusher.h"

#include <boost/foreach.hpp>
#include <boost/asio.hpp>
#include <boost/thread/thread.hpp>

#include <boost/log/trivial.hpp>
#include <boost/log/expressions.hpp>
#include <boost/log/utility/setup/file.hpp>
#include <boost/log/utility/setup/console.hpp>
#include <boost/log/utility/setup/common_attributes.hpp>
#include <boost/log/support/date_time.hpp>

using namespace std;

volatile int keepRunning;
pluginVector_t plugins;
boost::shared_ptr<boost::asio::io_service::work> keepAliveWork;

void sigintHandler(int sig) {
	boost::log::sources::severity_logger<boost::log::trivial::severity_level> lg;
	LOG(fatal) << "Received SIGINT";
	//Stop all sensors
	LOG(info) << "Stopping sensors...";
	for(auto& p : plugins) {
		LOG(info) << "Stop \"" << p.id << "\" plugin";
		for(auto s : p.configurator->getSensors()) {
			s->stopPolling();
		}
	}
	//Stop io service by killing keepAliveWork
	keepAliveWork.reset();
}

void sigtermHandler(int sig) {
	boost::log::sources::severity_logger<boost::log::trivial::severity_level> lg;
	LOG(fatal) << "Received SIGTERM. Terminating";
	signal(SIGTERM, SIG_DFL);
	kill(getpid(), SIGTERM);
}

void printSyntax()
{
/*
                       1         2         3         4         5         6         7         8
             012345678901234567890123456789012345678901234567890123456789012345678901234567890
*/
	cout << "Usage:" << endl;
	cout << "  dcdbpusher [-d] [-b<host>] [-p<port>] [-m<string>] [-t<number>] <path/to/configfiles/>" << endl;
	cout << "  dcdbpusher -h" << endl;
	cout << endl;

	cout << "Options:" << endl;
	cout << "  -b<host>        MQTT broker" << endl;
	cout << "  -p<port>        MQTT broker port" << endl;
	cout << "  -m<string>      MQTT topic prefix" << endl;
	cout << "  -t<number>      Thread count" << endl;
	cout << "  -w<path>        Writable directory for temporal files" << endl;
	cout << endl;
	cout << "  -d              Daemonize" << endl;
	cout << "  -v              Set verbosity of output." << endl
		 <<	"                  Can be a number between 5 (all output) and 0 (only fatal messages)." << endl
		 <<	"                  Default level is 3 (info)" << endl;
	cout << "  -h              This help page" << endl;
	cout << endl;
}

/*
 * Inverts the given severity level.
 * trace <=> fatal
 * debug <=> error
 * info <=> warning
 */
boost::log::trivial::severity_level invertLogLevel(boost::log::trivial::severity_level logLevel) {
	switch (logLevel) {
		case boost::log::trivial::trace:
			return boost::log::trivial::fatal;
			break;
		case boost::log::trivial::debug:
			return boost::log::trivial::error;
			break;
		case boost::log::trivial::info:
			return boost::log::trivial::warning;
			break;
		case boost::log::trivial::warning:
			return boost::log::trivial::info;
			break;
		case boost::log::trivial::error:
			return boost::log::trivial::debug;
			break;
		case boost::log::trivial::fatal:
			return boost::log::trivial::trace;
			break;
		default:
			return boost::log::trivial::info;
			break;
	}
}

int main(int argc, char** argv) {
	if (argc <= 1) {
		cout << "Please specify a path to the config-directory" << endl << endl;
		printSyntax();
		return 1;
	}

	//define allowed command-line options once
	const char opts[] = "b:p:m:t:v:w:dh";

	//check if help flag specified
	char c;
	while ((c = getopt(argc, argv, opts)) != -1) {
		switch (c)
		{
			case 'h':
				printSyntax();
				return 1;
				break;
			default:
				//do nothing (other options are read later on)
				break;
		}
	}

	//init LOGGING
	boost::log::add_common_attributes();

	//set up logger to command line
	auto cmdSink = boost::log::add_console_log(
		std::cout,
		boost::log::keywords::format = (boost::log::expressions::stream //only print timestamp (without date), severity and message to terminal
				<< "[" << boost::log::expressions::format_date_time< boost::posix_time::ptime >("TimeStamp", "%H:%M:%S") << "]"
				<< " <" << boost::log::trivial::severity << ">"
				<< ": " << boost::log::expressions::smessage
		),
		boost::log::keywords::filter = (boost::log::trivial::severity >= boost::log::trivial::info)
	);

	//get logger instance
	boost::log::sources::severity_logger<boost::log::trivial::severity_level> lg;
	//finished logging startup for the moment (file log added later)

	boost::asio::io_service io;
	boost::thread_group threads;
	Configuration cfg(argv[argc-1]);
	global_t globalSettings;
	keepRunning = 1;
	signal(SIGINT, sigintHandler);	//Handle Strg+C
	signal(SIGTERM, sigtermHandler);	//Handle termination

	//Read global variables from config file
	if(!cfg.readGlobal()) {
		LOG(fatal) << "Failed to read global configuration!";
		return 1;
	}
	globalSettings = cfg.getGlobal();

	//reset getopt()
	optind = 1;
	//read in options (overwrite global.conf settings if necessary)
	while ((c = getopt(argc, argv, opts)) != -1) {
		switch (c)
		{
			case 'b':
				globalSettings.brokerHost = optarg;
				break;
			case 'p':
				globalSettings.brokerPort = atoi(optarg);
				break;
			case 'm':
				globalSettings.mqttPrefix = optarg;
				break;
			case 't':
				globalSettings.threads = stoul(optarg);
				break;
			case 'v':
				globalSettings.logLevelCmd = static_cast<boost::log::trivial::severity_level>(stoi(optarg));
				break;
			case 'd':
				globalSettings.daemonize = 1;
				break;
			case 'w':
				globalSettings.tempdir = optarg;
				if (globalSettings.tempdir[globalSettings.tempdir.length()-1] != '/') {
					globalSettings.tempdir.append("/");
				}
				break;
			case 'h':
				printSyntax();
				return 1;
				break;
			default:
				if (c != '?') cerr << "Unknown parameter: " << c << endl;
				return 1;
		}
	}
	//propagate overwritten values back to cfg
	cfg.setGlobal(globalSettings);

	//we now should know where the writable tempdir is
	//set up logger to file
	auto fileSink = boost::log::add_file_log(
		boost::log::keywords::file_name = globalSettings.tempdir + "dcdb_%N.log",	//	number logfiles ascending
		boost::log::keywords::rotation_size = 10 * 1024 * 1024,	//rotate logfile every 10 MiB
		//boost::log::keywords::time_based_rotation = boost::log::sinks::file::rotation_at_time_point(0, 0, 0),	//Throws bad year-exception for no obvious reason
		boost::log::keywords::format = //	format:		LineID [Timestamp] ThreadID <severity>: Message
		(	boost::log::expressions::stream
				<< boost::log::expressions::attr< unsigned int >("LineID")
				<< " [" << boost::log::expressions::format_date_time< boost::posix_time::ptime >("TimeStamp", "%Y-%m-%d, %H:%M:%S") << "]"
				<< " " << boost::log::expressions::attr<boost::log::attributes::current_thread_id::value_type >("ThreadID")
				<< " <" << boost::log::trivial::severity << ">"
				<< ": " << boost::log::expressions::smessage
		),
		boost::log::keywords::filter = (boost::log::trivial::severity >= boost::log::trivial::trace),
		boost::log::keywords::auto_flush = true
	);

	//invert log level, because 0 casts to trace (and 5 to fatal) but we want the meaning to be the other way round (0 equals fatal, 5 trace)
	globalSettings.logLevelCmd = invertLogLevel(globalSettings.logLevelCmd);
	globalSettings.logLevelFile = invertLogLevel(globalSettings.logLevelFile);

	//severity level may be overwritten (per option or config-file) --> set it according to globalSettings
	fileSink->set_filter(boost::log::trivial::severity >= globalSettings.logLevelFile);
	cmdSink->set_filter(boost::log::trivial::severity >= globalSettings.logLevelCmd);

	LOG(info) << "Logging setup complete";

	//Read in rest of configuration. Also creates all sensors
	if(!cfg.read()) {
		LOG(fatal) << "Failed to read configuration!";
		return 1;
	}
	plugins = cfg.getPlugins();

	//give some feedback
	LOG(info) << "Global Settings:";
	LOG(info) << "  REST Server: " << globalSettings.restHost << ":" << globalSettings.restPort;
	LOG(info) << "  Broker:      " << globalSettings.brokerHost << ":" << globalSettings.brokerPort;
	LOG(info) << "  MQTT-prefix: " << globalSettings.mqttPrefix;
	LOG(info) << "  Threads:     " << globalSettings.threads;
	if(globalSettings.daemonize) {
		LOG(info) << "  Daemonize:   Enabled";
	} else {
		LOG(info) << "  Daemonize:   Disabled";
	}
	LOG(info) << "  Write-Dir:   " << globalSettings.tempdir;
	LOG(info) << "  CacheInterval: " << globalSettings.cacheInterval / 1000 << " [s]";

	//Init all sensors
	LOG(info) << "Init sensors...";
	for(auto& p : plugins) {
		LOG(info) << "Init \"" << p.id << "\" plugin";
		for(auto s : p.configurator->getSensors()) {
			LOG(debug) << " -" << s->getName();
			s->init(io);
		}
	}

	//Start all sensors
	LOG(info) << "Starting sensors...";
	for(auto& p : plugins) {
		LOG(info) << "Start \"" << p.id << "\" plugin";
		for(auto s : p.configurator->getSensors()) {
			s->startPolling();
		}
	}

	LOG(info) << "Sensors started!";

	if (globalSettings.daemonize) {
		//boost.log does not support forking officially.
		//however, just don't touch the sinks after daemonizing and it should work nonetheless
		LOG(info) << "Detaching...";

		cmdSink->flush();
		boost::log::core::get()->remove_sink(cmdSink);
		cmdSink.reset();

		//daemonize
		dcdbdaemon();

		LOG(info) << "Now detached";
	}

	LOG(info) << "Creating threads...";

	//dummy to keep io service alive even if no tasks remain (e.g. because all sensors have been stopped over REST API)
	keepAliveWork = boost::make_shared<boost::asio::io_service::work>(io);

	//Create pool of threads which handle the sensors
	for(size_t i = 0; i < globalSettings.threads; i++) {
		threads.create_thread(bind(static_cast< size_t (boost::asio::io_service::*) () >(&boost::asio::io_service::run), &io));
	}

	//MQTTPusher and Https server get their own threads
	MQTTPusher mqttPusher(globalSettings.brokerPort, globalSettings.brokerHost, globalSettings.mqttPrefix, plugins);
	HttpsServer httpsServer(globalSettings.restHost, globalSettings.restPort, plugins);

	boost::thread mqttThread(bind(&MQTTPusher::push, &mqttPusher));
	boost::thread restThread(bind(&HttpsServer::run, &httpsServer));

	LOG(info) << "Threads created!";
	LOG(info) << "Setup complete!";

	LOG(trace) << "Running...";

	//Run until Strg+C
	threads.join_all();

	//will only continue if interrupted by SIGINT and threads were stopped

	//Stop MQTTPusher
	LOG(info) << "Flushing MQTT queues...";
	keepRunning = 0;
	mqttThread.join();

	//Stop https server
	LOG(info) << "Stopping REST API Server...";
	httpsServer.stop();
	restThread.join();

	LOG(info) << "Exiting...Goodbye!";
	return 0;
}