Initial import of current chronos version

README.md

@@ -1,2 +1,43 @@
-# cron-job.org
-cron-job.org Open Source project
+cron-job.org
+============
+
+Structure
+---------
+* `database` contains the MySQL database structure.
+* `chronos` is cron-job.org's cron job execution daemon and is responsible for fetching the jobs.
+* `web` contains the web interface (coming soon)
+
+chronos
+-------
+### Concept
+chronos checks the MySQL database every minute to collect all jobs to execute. For every minute, a thread is spawned which processes all the jobs. Actual HTTP fetching is done using the excellent CURL multi library with libev as library used to provide the event loop. Together with the c-ares resovler this allows for thousands of parallel HTTP requests.
+
+cron-job.org supports storing the job results for the user's convenience. This can quickly lead to I/O bottleneck when storing the result data in a MySQL database. (Which also has the downside that cleaning up old entries is extremely expensive.) To solve this issue, chronos stores the results in per-user per-day SQLite databases. Cleaning up old entries is as easy as deleting the corresponding day's databases.
+
+The whole software is optimized on performance rather than on data integrity, i.e. when your server crashes or you have a power outage / hardware defect, the job history is most likely lost. Since this is volatile data anyway, it's not considered a big issue.
+
+### Prerequisites
+In order to build chronos, you need development files of:
+* curl (preferably with c-ares as resolver)
+* libev
+* mysqlclient
+* sqlite3
+To build, you need a C++14 compiler and cmake.
+
+### Building
+1. Create and enter a build folder: `mkdir build && cd build`
+2. Run cmake: `cmake -DCMAKE_BUILD_TYPE=Release ..`
+3. Build the project: `make`
+
+### Running
+1. Ensure you've imported the DB scheme from the `database` folder
+2. Customize `chronos.cfg` according to your system (especially add your MySQL login)
+3. Execute `./chronos /path/to/chronos.cfg`
+
+General notes
+-------------
+* Web interface and jitter correction algorithm are still missing in this repository and will be added as soon as they've been refactored to a presentable state.
+* We strongly recommend to build CURL using the c-ares resolver. Otherwise every request might spawn its own thread for DNS resolving and your machine will run out of resources *very* soon.
+* Before running chronos, ensure that the limit of open files/sockets is not set too low. You might want to run `ulimit -n 65536` or similar first.
+* If data integrity is not important for you, we highly recommend to set `innodb_flush_log_at_trx_commit=0` and `innodb_flush_method=O_DIRECT` in your MySQL config for best performance. Otherwise the update thread (which is responsible for storing the job resuls) might lag behind the actual job executions quite soon.
+* Parts of the source are quite old and from early stages of the project and might require a refactoring sooner or later.

chronos/App.cpp

@@ -0,0 +1,214 @@
+/*
+ * chronos, the cron-job.org execution daemon
+ * Copyright (C) 2017 Patrick Schlangen <[email protected]>
+ *
+ * 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.
+ *
+ */
+
+#include "App.h"
+
+#include <stdexcept>
+#include <iostream>
+#include <functional>
+
+#include <signal.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#include <unistd.h>
+
+#include <curl/curl.h>
+
+#include "UpdateThread.h"
+#include "WorkerThread.h"
+#include "Config.h"
+
+using namespace Chronos;
+
+App *App::instance = nullptr;
+
+App::App(int argc, char *argv[])
+{
+	if(App::instance != nullptr)
+		throw std::runtime_error("App instance already exists");
+
+	if(argc != 2)
+		throw std::runtime_error(std::string("Usage: ") + std::string(argv[0]) + std::string(" [config-file]"));
+
+	this->config 		= std::make_shared<Config>(argv[1]);
+	App::instance		= this;
+}
+
+App::~App()
+{
+	App::instance		= nullptr;
+}
+
+App *App::getInstance()
+{
+        if(App::instance == nullptr)
+                throw std::runtime_error("No app instance available");
+        return(App::instance);
+}
+
+void App::processJobs(int hour, int minute, int month, int mday, int wday, int year, time_t timestamp)
+{
+	std::cout 	<< "App::processJobs(): Called for "
+				<< "hour = " << hour << ", "
+				<< "minute = " << minute << ", "
+				<< "month = " << month << ", "
+				<< "mday = " << mday << ", "
+				<< "wday = " << wday << ", "
+				<< "timestamp = " << timestamp
+				<< std::endl;
+
+	auto res = db->query("SELECT TRIM(`url`),`job`.`jobid`,`auth_enable`,`auth_user`,`auth_pass`,`notify_failure`,`notify_success`,`notify_disable`,`fail_counter`,`save_responses`,`userid` FROM `job` "
+									"INNER JOIN `job_hours` ON `job_hours`.`jobid`=`job`.`jobid` "
+									"INNER JOIN `job_mdays` ON `job_mdays`.`jobid`=`job`.`jobid` "
+									"INNER JOIN `job_wdays` ON `job_wdays`.`jobid`=`job`.`jobid` "
+									"INNER JOIN `job_minutes` ON `job_minutes`.`jobid`=`job`.`jobid` "
+									"INNER JOIN `job_months` ON `job_months`.`jobid`=`job`.`jobid` "
+									"WHERE (`hour`=-1 OR `hour`=%d) "
+									"AND (`minute`=-1 OR `minute`=%d) "
+									"AND (`mday`=-1 OR `mday`=%d) "
+									"AND (`wday`=-1 OR `wday`=%d) "
+									"AND (`month`=-1 OR `month`=%d) "
+									"AND `enabled`=1 "
+									"ORDER BY `fail_counter` ASC, `last_duration` ASC",
+									hour, minute, mday, wday, month);
+
+	int jobCount = res->numRows();
+	std::cout << "App::processJobs(): " << jobCount << " jobs found" << std::endl;
+
+	if(jobCount > 0)
+	{
+		std::shared_ptr<WorkerThread> wt = std::make_shared<WorkerThread>(mday, month, year, hour, minute);
+
+		MYSQL_ROW row;
+		while((row = res->fetchRow()) != nullptr)
+		{
+			HTTPRequest *req = HTTPRequest::fromURL(row[0], atoi(row[10]), wt);
+			req->result->jobID 			= atoi(row[1]);
+			req->result->datePlanned	= (uint64_t)timestamp * 1000;
+			req->result->notifyFailure 	= strcmp(row[5], "1") == 0;
+			req->result->notifySuccess 	= strcmp(row[6], "1") == 0;
+			req->result->notifyDisable 	= strcmp(row[7], "1") == 0;
+			req->result->oldFailCounter	= atoi(row[8]);
+			req->result->saveResponses	= strcmp(row[9], "1") == 0;
+			if(atoi(row[2]) == 1)
+			{
+				req->useAuth 		= true;
+				req->authUsername 	= row[3];
+				req->authPassword 	= row[4];
+			}
+
+			wt->addJob(req);
+		}
+
+		std::cout << "App::processJobs(): Starting worker thread" << std::endl;
+		wt->run();
+	}
+
+	res.reset();
+
+	std::cout << "App::processJobs(): Finished" << std::endl;
+}
+
+void App::signalHandler(int sig)
+{
+	if(sig == SIGINT)
+		App::getInstance()->stop = true;
+}
+
+int App::run()
+{
+	curl_global_init(CURL_GLOBAL_ALL);
+	MySQL_DB::libInit();
+
+	db = createMySQLConnection();
+	startUpdateThread();
+
+	signal(SIGINT, App::signalHandler);
+
+	bool firstLoop = true;
+	struct tm lastTime = { 0 };
+	int jitterCorrectionOffset = calcJitterCorrectionOffset();
+	while(!stop)
+	{
+		time_t currentTime = time(nullptr) + jitterCorrectionOffset;
+		struct tm *t = localtime(&currentTime);
+
+		if(t->tm_min > lastTime.tm_min
+			|| t->tm_hour > lastTime.tm_hour
+			|| t->tm_mday > lastTime.tm_mday
+			|| t->tm_mon > lastTime.tm_mon
+			|| t->tm_year > lastTime.tm_year)
+		{
+			// update last time
+			memcpy(&lastTime, t, sizeof(struct tm));
+
+			if(!firstLoop || t->tm_sec == 59 - jitterCorrectionOffset)
+			{
+				processJobs(t->tm_hour, t->tm_min, t->tm_mon+1, t->tm_mday, t->tm_wday, t->tm_year+1900, currentTime - t->tm_sec);
+				jitterCorrectionOffset = calcJitterCorrectionOffset();
+			}
+
+			firstLoop = false;
+		}
+		else
+		{
+			usleep(100*1000);
+		}
+	}
+
+	this->stopUpdateThread();
+
+	MySQL_DB::libCleanup();
+	curl_global_cleanup();
+
+	return(1);
+}
+
+int App::calcJitterCorrectionOffset()
+{
+	return 1; //! @todo
+}
+
+void App::updateThreadMain()
+{
+	try
+	{
+		updateThreadObj = std::make_unique<UpdateThread>();
+		updateThreadObj->run();
+		updateThreadObj.reset();
+	}
+	catch(const std::runtime_error &ex)
+	{
+		std::cout << "Update thread runtime error: " << ex.what() << std::endl;
+		stop = true;
+	}
+}
+
+void App::startUpdateThread()
+{
+	updateThread = std::thread(std::bind(&App::updateThreadMain, this));
+}
+
+void App::stopUpdateThread()
+{
+	updateThreadObj->stopThread();
+	updateThread.join();
+}
+
+std::unique_ptr<MySQL_DB> App::createMySQLConnection()
+{
+	return(std::make_unique<MySQL_DB>(config->get("mysql_host"),
+						config->get("mysql_user"),
+						config->get("mysql_pass"),
+						config->get("mysql_db"),
+						config->get("mysql_sock")));
+}

chronos/App.h

@@ -0,0 +1,65 @@
+/*
+ * chronos, the cron-job.org execution daemon
+ * Copyright (C) 2017 Patrick Schlangen <[email protected]>
+ *
+ * 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.
+ *
+ */
+
+#ifndef _APP_H_
+#define _APP_H_
+
+#include <memory>
+#include <thread>
+
+#include <time.h>
+
+#include "MySQL.h"
+#include "Config.h"
+
+namespace Chronos
+{
+	class MySQL_DB;
+	class UpdateThread;
+
+	class App
+	{
+	public:
+		App(int argc, char *argv[]);
+		~App();
+
+	private:
+		App(const App &other) = delete;
+		App(App &&other) = delete;
+		App &operator=(const App &other) = delete;
+		App &operator=(App &&other) = delete;
+
+	public:
+		static App *getInstance();
+		static void signalHandler(int sig);
+		void updateThreadMain();
+		int run();
+		std::unique_ptr<MySQL_DB> createMySQLConnection();
+
+	private:
+		void startUpdateThread();
+		void stopUpdateThread();
+		void processJobs(int hour, int minute, int month, int mday, int wday, int year, time_t timestamp);
+		int calcJitterCorrectionOffset();
+
+	public:
+		std::shared_ptr<Config> config;
+
+	private:
+		bool stop = false;
+		static App *instance;
+		std::thread updateThread;
+		std::unique_ptr<MySQL_DB> db;
+		std::unique_ptr<UpdateThread> updateThreadObj;
+	};
+};
+
+#endif

chronos/CMake/Modules/FindMySQLClient.cmake

@@ -0,0 +1,7 @@
+find_path(MySQLClient_INCLUDE_DIRS mysql.h PATH_SUFFIXES mysql)
+
+find_library(MySQLClient_LIBRARIES NAMES libmysqlclient mysqlclient PATH_SUFFIXES mysql)
+
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(MySQLClient DEFAULT_MSG MySQLClient_LIBRARIES MySQLClient_INCLUDE_DIRS)
+mark_as_advanced(MySQLClient_INCLUDE_DIRS MySQLClient_LIBRARIES)

chronos/CMake/Modules/FindSQLite.cmake

@@ -0,0 +1,7 @@
+find_path(SQLITE_INCLUDE_DIRS sqlite3.h)
+
+find_library(SQLITE_LIBRARIES NAMES libsqlite3 sqlite3)
+
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(SQLite DEFAULT_MSG SQLITE_LIBRARIES SQLITE_INCLUDE_DIRS)
+mark_as_advanced(SQLITE_INCLUDE_DIRS SQLITE_LIBRARIES)

chronos/CMake/Modules/Findlibev.cmake

@@ -0,0 +1,7 @@
+find_path(LIBEV_INCLUDE_DIRS ev.h)
+
+find_library(LIBEV_LIBRARIES NAMES libev ev)
+
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(libev DEFAULT_MSG LIBEV_LIBRARIES LIBEV_INCLUDE_DIRS)
+mark_as_advanced(LIBEV_INCLUDE_DIRS LIBEV_LIBRARIES)

chronos/CMakeLists.txt

@@ -0,0 +1,45 @@
+cmake_minimum_required(VERSION 3.1)
+project(chronos)
+
+set(CMAKE_MODULE_PATH ${CMAKE_MODULE_PATH} "${CMAKE_SOURCE_DIR}/CMake/Modules")
+
+set(CMAKE_CXX_STANDARD 14)
+
+set(SOURCES
+	main.cpp
+	App.cpp
+	Config.cpp
+	HTTPRequest.cpp
+	MySQL_DB.cpp
+	MySQL_Result.cpp
+	SQLite.cpp
+	UpdateThread.cpp
+	Utils.cpp
+	WorkerThread.cpp
+)
+
+add_executable(${PROJECT_NAME} ${SOURCES})
+
+find_package(MySQLClient REQUIRED)
+find_package(CURL REQUIRED)
+find_package(libev REQUIRED)
+find_package(SQLite REQUIRED)
+find_package(Threads REQUIRED)
+
+include_directories(${MySQLClient_INCLUDE_DIRS}
+	${CURL_INCLUDE_DIRS}
+	${LIBEV_INCLUDE_DIRS}
+	${SQLITE_INCLUDE_DIRS}
+)
+target_link_libraries(${PROJECT_NAME}
+	${MySQLClient_LIBRARIES}
+	${CURL_LIBRARIES}
+	${LIBEV_LIBRARIES}
+	${SQLITE_LIBRARIES}
+	${CMAKE_THREAD_LIBS_INIT}
+)
+
+include_directories(${PROJECT_SOURCE_DIR})
+
+install(TARGETS ${PROJECT_NAME} DESTINATION bin)
+