redesign announce jitter to better spread the load after startup

output user report and info_torrent as JSON

See merge request tracker/ocelot!7

.dockerignore

@@ -0,0 +1,3 @@
+.vscode/
+build/
+/ocelot

.gitignore

@@ -1,4 +1,5 @@
 .idea/
+.vscode/
 .vagrant/
 *.tar.*
 MakeFile.in

CHANGES

@@ -1,3 +1,21 @@
+-- 2.1.0 (2024-02-11)
+FEATURES
+* announce jitter is now a uniform distribution and can be modified on the fly
+
+-- 2.0.1 (2024-02-10)
+BREAKING
+* API torrent endpoint output changed to JSON
+* API user endpoint shows more information about a user in JSON format
+FEATURES
+* Mysql port may be specified in configuration file
+* API stats report contains Ocelot version
+* Queue sizes are reported in stats
+* Sleep if Gazelle API is down
+* Handle DB connection problems gracefully
+* Add Prometheus and Grafana support
+BUILD
+* Docker image based on Bookworm
+
 -- 2.0.0 (2019-10-22)
 BREAKING
 * Write leech stats into their own dedicated tables (users_leech_stats, torrents_leech_stats)
@@ -15,7 +33,7 @@ BUILD
 Move to spdlog (https://github.com/gabime/spdlog)to handle logging instead of cout
     spdlog is vendored under src/, currently using v0.13.0
 Add logging directive to config to configure how Ocelot logs and where it logs
-Add deamon (--daemon, -d) to run Ocelot as a daemon
+Add daemon switch (--daemon, -d) to run Ocelot as a daemon
 Move source code into src/ directory
 Add Vagrantfile and Dockerfile for running Ocelot
 

CMakeLists.txt

@@ -12,7 +12,7 @@ endif()
 list(APPEND CMAKE_MODULE_PATH "${CMAKE_CURRENT_LIST_DIR}/CMake")
 set(CMAKE_CXX_STANDARD 11)
 set(CMAKE_CXX_STANDARD_REQUIRED ON)
-set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11 -march=native -O2 -fvisibility=hidden -fvisibility-inlines-hidden -fomit-frame-pointer -fno-ident -Wall -Wfatal-errors")
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -fvisibility=hidden -fvisibility-inlines-hidden -Wall -Wfatal-errors")
 
 message(STATUS "CMAKE_MODULE_PATH='${CMAKE_MODULE_PATH}'")
 find_package(Jemalloc)
@@ -66,12 +66,10 @@ SET(LINK_LIBRARIES
 
 if (USE_JEMALLOC)
     message(STATUS "Build with jemalloc")
-    list(APPEND ${LINK_LIBRARIES} ${JEMALLOC_LIBRARY})
-    SET(CMAKE_EXE_LINKER_FLAGS  "${CMAKE_EXE_LINKER_FLAGS} -ljemalloc")
+    list(APPEND LINK_LIBRARIES ${JEMALLOC_LIBRARY})
 elseif (USE_TCMALLOC)
     message(STATUS "Build with tcmalloc")
-    list(APPEND ${LINK_LIBRARIES} ${TCMALLOC_LIBRARY})
-    SET(CMAKE_EXE_LINKER_FLAGS  "${CMAKE_EXE_LINKER_FLAGS} -ltcmalloc")
+    list(APPEND LINK_LIBRARIES ${TCMALLOC_LIBRARY})
 endif()
 
 target_link_libraries(

Dockerfile

@@ -1,7 +1,6 @@
-FROM debian:stretch-slim
+FROM debian:bookworm-slim
 
-COPY . /srv
-WORKDIR /srv
+LABEL org.opencontainers.image.source=https://github.com/OPSnet/Ocelot
 
 RUN apt-get update \
     && apt-get install --no-install-recommends -y \
@@ -13,19 +12,25 @@ RUN apt-get update \
         libev-dev \
         libjemalloc-dev \
         libmysql++-dev \
+        netcat-traditional \
         pkg-config \
-    && mkdir build \
-    && cd build \
-    && cmake .. \
-    && make \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY . /srv
+WORKDIR /srv
+
+RUN cmake -Wno-dev -S /srv -B /srv/build \
+    && make -C build \
     && apt-get purge -y \
         build-essential \
-        cmake \    
+        cmake \
         pkg-config \
     && apt-get autoremove -y \
     && apt-get clean -y \
-    && rm -rf /var/lib/apt/lists/* \
     && mv /srv/build/ocelot /srv/ocelot \
     && mv /srv/ocelot.conf.dist /srv/ocelot.conf
 
-CMD ["/srv/ocelot"]
+# default listen_port value in ocelot.conf
+EXPOSE 34000
+
+CMD ["/srv/ocelot"]

README.md

@@ -5,17 +5,16 @@ It supports requests over TCP and can only track IPv4 peers.
 
 ## Ocelot Compile-time Dependencies
 
-* [GCC/G++](http://gcc.gnu.org/) (4.7+ required; 4.8.1+ recommended)
-* [Boost](http://www.boost.org/) (1.55.0+ required)
+* [GCC/G++](http://gcc.gnu.org/) (11+ required; 11.4.0+ recommended)
+* [Boost](http://www.boost.org/) (1.74.0+ required)
 * [libev](http://software.schmorp.de/pkg/libev.html) (required)
-* [spdlog](https://github.com/gabime/spdlog) (required)
 * [MySQL++](http://tangentsoft.net/mysql++/) (3.2.0+ required)
 * [jemalloc](http://jemalloc.net/) (optional, but strongly recommended)
 * [TCMalloc](http://goog-perftools.sourceforge.net/doc/tcmalloc.html) (optional)
 
 ## Installation
 
-### Debian Stretch
+### Debian Bookworm
 ```bash
 sudo apt-get install \
     build-essential \
@@ -26,17 +25,27 @@ sudo apt-get install \
     libev-dev \
     libjemalloc-dev \
     libmysql++-dev \
+	netcat-traditional \
     pkg-config
-mkdir build
-cd build
-cmake ..
-make
+cmake -Wno-dev . -B build 
+make -C build
 ```
 
 The [Gazelle installation guides](https://github.com/WhatCD/Gazelle/wiki/Gazelle-installation) include instructions for installing Ocelot as a part of the Gazelle project.
 
+### Docker
+
+```bash
+docker build . -t ocelot
+docker run -v $(pwd)/ocelot.conf:/srv/ocelot.conf ocelot
+```
+
 ### Standalone Installation
 
+* Prepare the build environment. (This must be re-executed when new source files are added).
+
+        cmake -Wno-dev . -B build
+
 * Create the following tables according to the [Gazelle database schema](https://raw.githubusercontent.com/WhatCD/Gazelle/master/gazelle.sql):
  - `torrents`
  - `torrents_leech_stats`
@@ -51,10 +60,7 @@ The [Gazelle installation guides](https://github.com/WhatCD/Gazelle/wiki/Gazelle
 
 * Build Ocelot:
 
-        mkdir build
-        cd build
-        cmake ..
-        make
+        make -C build
 
 ## Running Ocelot
 

ocelot.conf.dist

@@ -11,14 +11,16 @@ connection_timeout  = 10
 keepalive_timeout   = 0
 
 announce_interval   = 1800
+announce_jitter     = 240
 max_request_size    = 4096
 numwant_limit       = 50
 request_log_size    = 500
 
-mysql_host          = 
-mysql_username      = 
-mysql_password      = 
-mysql_db            = 
+mysql_host          = mysql
+mysql_port          = 36000
+mysql_username      = gazelle
+mysql_password      = password
+mysql_db            = gazelle
 
 # The passwords must be 32 characters and match the Gazelle config
 report_password     = 00000000000000000000000000000000

src/config.cpp

@@ -66,6 +66,7 @@ void config::init() {
 
 	// Tracker requests
 	add("announce_interval", 1800u);
+	add("announce_jitter", 240u);
 	add("max_request_size", 4096u);
 	add("numwant_limit", 50u);
 	add("request_log_size", 500u);
@@ -79,8 +80,10 @@ void config::init() {
 	// MySQL
 	add("mysql_db", "gazelle");
 	add("mysql_host", "localhost");
+	add("mysql_port", 3306u);
 	add("mysql_username", "");
 	add("mysql_password", "");
+	add("mysql_port", 3306u);
 
 	// Site communication
 	add("site_host", "127.0.0.1");
@@ -100,7 +103,7 @@ void config::init() {
 confval * config::get(const std::string &setting_name) {
 	const auto setting = settings.find(setting_name);
 	if (setting == settings.end()) {
-		 spdlog::get("logger")->info("WARNING: Unrecognized setting '" + setting_name + "'");
+		spdlog::get("logger")->info("WARNING: Unrecognized setting '" + setting_name + "'");
 		return dummy_setting;
 	}
 	return &setting->second;

src/db.cpp

@@ -3,6 +3,7 @@
 #include "user.h"
 #include "misc_functions.h"
 #include "config.h"
+#include <chrono>
 #include <string>
 #include <iostream>
 #include <queue>
@@ -18,7 +19,7 @@
 mysql::mysql(config * conf) : u_active(false), t_active(false), p_active(false), s_active(false), tok_active(false) {
 	logger = spdlog::get("logger");
 	load_config(conf);
-	if (mysql_db == "") {
+	if (mysql_db.empty()) {
 		logger->error("No database selected");
 		return;
 	}
@@ -26,7 +27,7 @@ mysql::mysql(config * conf) : u_active(false), t_active(false), p_active(false),
 	try {
 		mysqlpp::ReconnectOption reconnect(true);
 		conn.set_option(&reconnect);
-		conn.connect(mysql_db.c_str(), mysql_host.c_str(), mysql_username.c_str(), mysql_password.c_str(), 0);
+		conn.connect(mysql_db.c_str(), mysql_host.c_str(), mysql_username.c_str(), mysql_password.c_str(), mysql_port);
 	} catch (const mysqlpp::Exception &er) {
 		logger->error("Failed to connect to MySQL (" + std::string(er.what()) + ")");
 		return;
@@ -45,6 +46,7 @@ void mysql::load_config(config * conf) {
 	mysql_host = conf->get_str("mysql_host");
 	mysql_username = conf->get_str("mysql_username");
 	mysql_password = conf->get_str("mysql_password");
+	mysql_port = conf->get_uint("mysql_port");
 	readonly = conf->get_bool("readonly");
 }
 
@@ -56,6 +58,12 @@ bool mysql::connected() {
 	return conn.connected();
 }
 
+mysqlpp::Connection mysql::create_connection() {
+	mysqlpp::Connection c(conn);
+	c.set_option(new mysqlpp::ReconnectOption(true));
+	return c;
+}
+
 void mysql::clear_peer_data() {
 	try {
 		mysqlpp::Query query = conn.query("TRUNCATE xbt_files_users;");
@@ -92,7 +100,7 @@ void mysql::load_torrents(torrent_list &torrents) {
 		for (size_t i = 0; i < num_rows; i++) {
 			std::string info_hash;
 			res[i][1].to_string(info_hash);
-			if (info_hash == "") {
+			if (info_hash.empty()) {
 				continue;
 			}
 			mysqlpp::sql_enum free_torrent(res[i][2]);
@@ -234,14 +242,14 @@ void mysql::load_whitelist(std::vector<std::string> &whitelist) {
 }
 
 void mysql::record_token(const std::string &record) {
-	if (update_token_buffer != "") {
+	if (!update_token_buffer.empty()) {
 		update_token_buffer += ",";
 	}
 	update_token_buffer += record;
 }
 
 void mysql::record_user(const std::string &record) {
-	if (update_user_buffer != "") {
+	if (!update_user_buffer.empty()) {
 		update_user_buffer += ",";
 	}
 	update_user_buffer += record;
@@ -249,14 +257,14 @@ void mysql::record_user(const std::string &record) {
 
 void mysql::record_torrent(const std::string &record) {
 	std::lock_guard<std::mutex> tb_lock(torrent_buffer_lock);
-	if (update_torrent_buffer != "") {
+	if (!update_torrent_buffer.empty()) {
 		update_torrent_buffer += ",";
 	}
 	update_torrent_buffer += record;
 }
 
 void mysql::record_peer(const std::string &record, const std::string &ip, const std::string &peer_id, const std::string &useragent) {
-	if (update_heavy_peer_buffer != "") {
+	if (!update_heavy_peer_buffer.empty()) {
 		update_heavy_peer_buffer += ",";
 	}
 	mysqlpp::Query q = conn.query();
@@ -265,7 +273,7 @@ void mysql::record_peer(const std::string &record, const std::string &ip, const
 	update_heavy_peer_buffer += q.str();
 }
 void mysql::record_peer(const std::string &record, const std::string &peer_id) {
-	if (update_light_peer_buffer != "") {
+	if (!update_light_peer_buffer.empty()) {
 		update_light_peer_buffer += ",";
 	}
 	mysqlpp::Query q = conn.query();
@@ -275,7 +283,7 @@ void mysql::record_peer(const std::string &record, const std::string &peer_id) {
 }
 
 void mysql::record_snatch(const std::string &record, const std::string &ip) {
-	if (update_snatch_buffer != "") {
+	if (!update_snatch_buffer.empty()) {
 		update_snatch_buffer += ",";
 	}
 	mysqlpp::Query q = conn.query();
@@ -306,13 +314,14 @@ void mysql::flush_users() {
 	if (verbose_flush || qsize > 0) {
 		logger->info("User flush queue size: " + std::to_string(qsize) + ", next query length: " + std::to_string(user_queue.front().size()));
 	}
-	if (update_user_buffer == "") {
+	if (update_user_buffer.empty()) {
 		return;
 	}
 	sql = "INSERT INTO users_leech_stats (UserID, Uploaded, Downloaded) VALUES " + update_user_buffer +
 		" ON DUPLICATE KEY UPDATE Uploaded = Uploaded + VALUES(Uploaded), Downloaded = Downloaded + VALUES(Downloaded)";
 	user_queue.push(sql);
 	update_user_buffer.clear();
+	stats.user_queue_size = qsize + 1;
 	if (!u_active) {
 		std::thread thread(&mysql::do_flush_users, this);
 		thread.detach();
@@ -331,7 +340,7 @@ void mysql::flush_torrents() {
 	if (verbose_flush || qsize > 0) {
 		logger->info("Torrent flush queue size: " + std::to_string(qsize) + ", next query length: " + std::to_string(torrent_queue.front().size()));
 	}
-	if (update_torrent_buffer == "") {
+	if (update_torrent_buffer.empty()) {
 		return;
 	}
 	sql = "INSERT INTO torrents_leech_stats (TorrentID,Seeders,Leechers,Snatched,Balance) VALUES " + update_torrent_buffer +
@@ -343,6 +352,7 @@ void mysql::flush_torrents() {
 	sql.clear();
 	sql = "DELETE FROM torrents WHERE info_hash = ''";
 	torrent_queue.push(sql);
+	stats.torrent_queue_size = qsize + 2;
 	if (!t_active) {
 		std::thread thread(&mysql::do_flush_torrents, this);
 		thread.detach();
@@ -360,12 +370,13 @@ void mysql::flush_snatches() {
 	if (verbose_flush || qsize > 0) {
 		logger->info("Snatch flush queue size: " + std::to_string(qsize) + ", next query length: " + std::to_string(snatch_queue.front().size()));
 	}
-	if (update_snatch_buffer == "" ) {
+	if (update_snatch_buffer.empty()) {
 		return;
 	}
-	sql = "INSERT INTO xbt_snatched (uid, fid, tstamp, IP) VALUES " + update_snatch_buffer;
+	sql = "INSERT IGNORE INTO xbt_snatched (uid, fid, tstamp, IP) VALUES " + update_snatch_buffer;
 	snatch_queue.push(sql);
 	update_snatch_buffer.clear();
+	stats.snatch_queue_size = qsize + 1;
 	if (!s_active) {
 		std::thread thread(&mysql::do_flush_snatches, this);
 		thread.detach();
@@ -381,22 +392,25 @@ void mysql::flush_peers() {
 	std::string sql;
 	std::lock_guard<std::mutex> pq_lock(peer_queue_lock);
 	size_t qsize = peer_queue.size();
+	short qsize_added = 0;
 	if (verbose_flush || qsize > 0) {
 		logger->info("Peer flush queue size: " + std::to_string(qsize) + ", next query length: " + std::to_string(peer_queue.front().size()));
 	}
 
 	// Nothing to do
-	if (update_light_peer_buffer == "" && update_heavy_peer_buffer == "") {
+	if (update_light_peer_buffer.empty() && update_heavy_peer_buffer.empty()) {
 		return;
 	}
 
-	if (update_heavy_peer_buffer != "") {
+	if (!update_heavy_peer_buffer.empty()) {
 		// Because xfu inserts are slow and ram is not infinite we need to
 		// limit this queue's size
 		// xfu will be messed up if the light query inserts a new row,
 		// but that's better than an oom crash
 		if (qsize >= 1000) {
 			peer_queue.pop();
+		} else {
+			qsize_added += 1;
 		}
 		sql = "INSERT INTO xbt_files_users (uid,fid,active,uploaded,downloaded,upspeed,downspeed,remaining,corrupt," +
 			std::string("timespent,announced,ip,peer_id,useragent,mtime) VALUES ") + update_heavy_peer_buffer +
@@ -409,10 +423,12 @@ void mysql::flush_peers() {
 		update_heavy_peer_buffer.clear();
 		sql.clear();
 	}
-	if (update_light_peer_buffer != "") {
+	if (!update_light_peer_buffer.empty()) {
 		// See comment above
 		if (qsize >= 1000) {
 			peer_queue.pop();
+		} else {
+			qsize_added += 1;
 		}
 		sql = "INSERT INTO xbt_files_users (uid,fid,timespent,announced,peer_id,mtime) VALUES " +
 					update_light_peer_buffer +
@@ -422,6 +438,7 @@ void mysql::flush_peers() {
 		update_light_peer_buffer.clear();
 		sql.clear();
 	}
+	stats.peer_queue_size = qsize + qsize_added;
 
 	if (!p_active) {
 		std::thread thread(&mysql::do_flush_peers, this);
@@ -440,13 +457,14 @@ void mysql::flush_tokens() {
 	if (verbose_flush || qsize > 0) {
 		logger->info("Token flush queue size: " + std::to_string(qsize) + ", next query length: " + std::to_string(token_queue.front().size()));
 	}
-	if (update_token_buffer == "") {
+	if (update_token_buffer.empty()) {
 		return;
 	}
 	sql = "INSERT INTO users_freeleeches (UserID, TorrentID, Downloaded) VALUES " + update_token_buffer +
 		" ON DUPLICATE KEY UPDATE Downloaded = Downloaded + VALUES(Downloaded)";
 	token_queue.push(sql);
 	update_token_buffer.clear();
+	stats.token_queue_size = qsize + 1;
 	if (!tok_active) {
 		std::thread thread(&mysql::do_flush_tokens, this);
 		thread.detach();
@@ -456,7 +474,7 @@ void mysql::flush_tokens() {
 void mysql::do_flush_users() {
 	u_active = true;
 	try {
-		mysqlpp::Connection c(mysql_db.c_str(), mysql_host.c_str(), mysql_username.c_str(), mysql_password.c_str(), 0);
+		mysqlpp::Connection c = create_connection();
 		while (user_queue.size() > 0) {
 			try {
 				std::string sql = user_queue.front();
@@ -464,20 +482,21 @@ void mysql::do_flush_users() {
 				if (!query.exec()) {
 					logger->info("User flush failed (" + std::to_string(user_queue.size()) + " remain)");
 					sleep(3);
-					continue;
+					break;
 				} else {
 					std::lock_guard<std::mutex> uq_lock(user_queue_lock);
 					user_queue.pop();
+					stats.user_queue_size -= 1;
 				}
 			}
 			catch (const mysqlpp::BadQuery &er) {
 				logger->error("Query error: " + std::string(er.what()) + " in flush users with a qlength: " + std::to_string(user_queue.front().size()) + " queue size: " + std::to_string(user_queue.size()));
-				sleep(3);
-				continue;
+				std::this_thread::sleep_for(std::chrono::seconds(3));
+				break;
 			} catch (const mysqlpp::Exception &er) {
 				logger->error("Query error: " + std::string(er.what()) + " in flush users with a qlength: " + std::to_string(user_queue.front().size()) +  " queue size: " + std::to_string(user_queue.size()));
-				sleep(3);
-				continue;
+				std::this_thread::sleep_for(std::chrono::seconds(3));
+				break;
 			}
 		}
 	}
@@ -490,32 +509,34 @@ void mysql::do_flush_users() {
 void mysql::do_flush_torrents() {
 	t_active = true;
 	try {
-		mysqlpp::Connection c(mysql_db.c_str(), mysql_host.c_str(), mysql_username.c_str(), mysql_password.c_str(), 0);
+		mysqlpp::Connection c = create_connection();
 		while (torrent_queue.size() > 0) {
 			try {
 				std::string sql = torrent_queue.front();
-				if (sql == "") {
+				if (sql.empty()) {
 					torrent_queue.pop();
+					stats.torrent_queue_size -= 1;
 					continue;
 				}
 				mysqlpp::Query query = c.query(sql);
 				if (!query.exec()) {
 					logger->info("Torrent flush failed (" + std::to_string(torrent_queue.size()) + " remain)");
-					sleep(3);
-					continue;
+					std::this_thread::sleep_for(std::chrono::seconds(3));
+					break;
 				} else {
 					std::lock_guard<std::mutex> tq_lock(torrent_queue_lock);
 					torrent_queue.pop();
+					stats.torrent_queue_size -= 1;
 				}
 			}
 			catch (const mysqlpp::BadQuery &er) {
 				logger->error("Query error: " + std::string(er.what()) + " in flush torrents with a qlength: " + std::to_string(torrent_queue.front().size()) + " queue size: " + std::to_string(torrent_queue.size()));
-				sleep(3);
-				continue;
+				std::this_thread::sleep_for(std::chrono::seconds(3));
+				break;
 			} catch (const mysqlpp::Exception &er) {
 				logger->error("Query error: " + std::string(er.what()) + " in flush torrents with a qlength: " + std::to_string(torrent_queue.front().size()) + " queue size: " + std::to_string(torrent_queue.size()));
-				sleep(3);
-				continue;
+				std::this_thread::sleep_for(std::chrono::seconds(3));
+				break;
 			}
 		}
 	}
@@ -528,28 +549,29 @@ void mysql::do_flush_torrents() {
 void mysql::do_flush_peers() {
 	p_active = true;
 	try {
-		mysqlpp::Connection c(mysql_db.c_str(), mysql_host.c_str(), mysql_username.c_str(), mysql_password.c_str(), 0);
+		mysqlpp::Connection c = create_connection();
 		while (peer_queue.size() > 0) {
 			try {
 				std::string sql = peer_queue.front();
 				mysqlpp::Query query = c.query(sql);
 				if (!query.exec()) {
 					logger->info("Peer flush failed (" + std::to_string(peer_queue.size()) + " remain)");
-					sleep(3);
-					continue;
+					std::this_thread::sleep_for(std::chrono::seconds(3));
+					break;
 				} else {
 					std::lock_guard<std::mutex> pq_lock(peer_queue_lock);
 					peer_queue.pop();
+					stats.peer_queue_size -= 1;
 				}
 			}
 			catch (const mysqlpp::BadQuery &er) {
 				logger->error("Query error: " + std::string(er.what()) + " in flush peers with a qlength: " + std::to_string(peer_queue.front().size()) + " queue size: " + std::to_string(peer_queue.size()));
-				sleep(3);
-				continue;
+				std::this_thread::sleep_for(std::chrono::seconds(3));
+				break;
 			} catch (const mysqlpp::Exception &er) {
 				logger->error("Query error: " + std::string(er.what()) + " in flush peers with a qlength: " + std::to_string(peer_queue.front().size()) + " queue size: " + std::to_string(peer_queue.size()));
-				sleep(3);
-				continue;
+				std::this_thread::sleep_for(std::chrono::seconds(3));
+				break;
 			}
 		}
 	}
@@ -562,28 +584,29 @@ void mysql::do_flush_peers() {
 void mysql::do_flush_snatches() {
 	s_active = true;
 	try {
-		mysqlpp::Connection c(mysql_db.c_str(), mysql_host.c_str(), mysql_username.c_str(), mysql_password.c_str(), 0);
+		mysqlpp::Connection c = create_connection();
 		while (snatch_queue.size() > 0) {
 			try {
 				std::string sql = snatch_queue.front();
 				mysqlpp::Query query = c.query(sql);
 				if (!query.exec()) {
 					logger->info("Snatch flush failed (" + std::to_string(snatch_queue.size()) + " remain)");
-					sleep(3);
-					continue;
+					std::this_thread::sleep_for(std::chrono::seconds(3));
+					break;
 				} else {
 					std::lock_guard<std::mutex> sq_lock(snatch_queue_lock);
 					snatch_queue.pop();
+					stats.snatch_queue_size -= 1;
 				}
 			}
 			catch (const mysqlpp::BadQuery &er) {
 				logger->error("Query error: " + std::string(er.what()) + " in flush snatches with a qlength: " + std::to_string(snatch_queue.front().size()) + " queue size: " + std::to_string(snatch_queue.size()));
-				sleep(3);
-				continue;
+				std::this_thread::sleep_for(std::chrono::seconds(3));
+				break;
 			} catch (const mysqlpp::Exception &er) {
 				logger->error("Query error: " + std::string(er.what()) + " in flush snatches with a qlength: " + std::to_string(snatch_queue.front().size()) + " queue size: " + std::to_string(snatch_queue.size()));
-				sleep(3);
-				continue;
+				std::this_thread::sleep_for(std::chrono::seconds(3));
+				break;
 			}
 		}
 	}
@@ -596,28 +619,29 @@ void mysql::do_flush_snatches() {
 void mysql::do_flush_tokens() {
 	tok_active = true;
 	try {
-		mysqlpp::Connection c(mysql_db.c_str(), mysql_host.c_str(), mysql_username.c_str(), mysql_password.c_str(), 0);
+		mysqlpp::Connection c = create_connection();
 		while (token_queue.size() > 0) {
 			try {
 				std::string sql = token_queue.front();
 				mysqlpp::Query query = c.query(sql);
 				if (!query.exec()) {
 					logger->info("Token flush failed (" + std::to_string(token_queue.size()) + " remain)");
-					sleep(3);
-					continue;
+					std::this_thread::sleep_for(std::chrono::seconds(3));
+					break;
 				} else {
 					std::lock_guard<std::mutex> tq_lock(token_queue_lock);
 					token_queue.pop();
+					stats.token_queue_size -= 1;
 				}
 			}
 			catch (const mysqlpp::BadQuery &er) {
 				logger->error("Query error: " + std::string(er.what()) + " in flush tokens with a qlength: " + std::to_string(token_queue.front().size()) + " queue size: " + std::to_string(token_queue.size()));
-				sleep(3);
-				continue;
+				std::this_thread::sleep_for(std::chrono::seconds(3));
+				break;
 			} catch (const mysqlpp::Exception &er) {
 				logger->error("Query error: " + std::string(er.what()) + " in flush tokens with a qlength: " + std::to_string(token_queue.front().size()) + " queue size: " + std::to_string(token_queue.size()));
-				sleep(3);
-				continue;
+				std::this_thread::sleep_for(std::chrono::seconds(3));
+				break;
 			}
 		}
 	}

src/db.h

@@ -26,6 +26,7 @@ class mysql {
 		std::queue<std::string> token_queue;
 
 		std::string mysql_db, mysql_host, mysql_username, mysql_password;
+		unsigned int mysql_port;
 		bool u_active, t_active, p_active, s_active, tok_active;
 		bool readonly;
 
@@ -42,6 +43,7 @@ class mysql {
 
 		void load_config(config * conf);
 		void load_tokens(torrent_list &torrents);
+		mysqlpp::Connection create_connection();
 
 		void do_flush_users();
 		void do_flush_torrents();

src/ocelot.cpp

@@ -22,6 +22,11 @@ static schedule *sched;
 
 struct stats_t stats;
 
+const char * version()
+{
+	return "2.1.0";
+}
+
 static void create_daemon()
 {
 	pid_t pid;
@@ -92,7 +97,9 @@ int main(int argc, char **argv) {
 
 	conf = new config();
 
-	bool verbose = false, conf_arg = false, daemonize = false;
+	bool conf_arg  = false;
+	bool daemonize = false;
+	bool verbose   = false;
 	std::string conf_file_path("./ocelot.conf");
 
 	for (int i = 1; i < argc; i++) {
@@ -107,11 +114,11 @@ int main(int argc, char **argv) {
 			conf_file_path = argv[++i];
 		}
 		else if(strcmp(argv[i], "-V") == 0 || strcmp(argv[i], "--version") == 0) {
-			std::cout << "Ocelot, version 2.0.0" << std::endl;
+			std::cout << "Ocelot version " << version() << std::endl;
 			return 0;
 		}
 		else {
-			std::cout << "Usage: " << argv[0] << " [-v] [-c configfile]" << std::endl;
+			std::cout << "Usage: " << argv[0] << " [-v] [-c configfile] [--daemonize]" << std::endl;
 			return 0;
 		}
 	}

src/ocelot.h

@@ -95,6 +95,11 @@ struct stats_t {
 	std::atomic<uint64_t> connection_rate;
 	std::atomic<uint32_t> leechers;
 	std::atomic<uint32_t> seeders;
+	std::atomic<uint32_t> user_queue_size;
+	std::atomic<uint32_t> torrent_queue_size;
+	std::atomic<uint32_t> peer_queue_size;
+	std::atomic<uint32_t> snatch_queue_size;
+	std::atomic<uint32_t> token_queue_size;
 	std::atomic<uint64_t> requests;
 	std::atomic<uint64_t> request_rate;
 	std::atomic<uint64_t> announcements;
@@ -105,4 +110,6 @@ struct stats_t {
 	time_t start_time;
 };
 extern struct stats_t stats;
+
+extern const char *version();
 #endif

src/report.cpp

@@ -7,12 +7,10 @@
 #include "response.h"
 #include "user.h"
 
-std::string report(params_type &params, user_list &users_list, client_opts_t &client_opts) {
+std::string report(params_type &params, user_list &users_list, client_opts_t &client_opts, unsigned int announce_interval, unsigned int announce_jitter) {
 	std::stringstream output;
 	std::string action = params["get"];
-	if (action == "") {
-		output << "Invalid action\n";
-	} else if (action == "stats") {
+	if (action == "stats") {
 		time_t uptime = time(NULL) - stats.start_time;
 		int up_d = uptime / 86400;
 		uptime -= up_d * 86400;
@@ -20,23 +18,51 @@ std::string report(params_type &params, user_list &users_list, client_opts_t &cl
 		uptime -= up_h * 3600;
 		int up_m = uptime / 60;
 		int up_s = uptime - up_m * 60;
-		std::string up_ht = up_h <= 9 ? '0' + inttostr(up_h) : inttostr(up_h);
-		std::string up_mt = up_m <= 9 ? '0' + inttostr(up_m) : inttostr(up_m);
-		std::string up_st = up_s <= 9 ? '0' + inttostr(up_s) : inttostr(up_s);
 
-		output << "Uptime: " << up_d << " days, " << up_ht << ':' << up_mt << ':' << up_st << '\n'
-			   << stats.opened_connections << " connections opened\n"
-			   << stats.open_connections << " open connections\n"
-			   << stats.connection_rate << " connections/s\n"
-			   << stats.requests << " requests handled\n"
-			   << stats.request_rate << " requests/s\n"
-			   << stats.succ_announcements << " successful announcements\n"
-			   << (stats.announcements - stats.succ_announcements) << " failed announcements\n"
-			   << stats.scrapes << " scrapes\n"
-			   << stats.leechers << " leechers tracked\n"
-			   << stats.seeders << " seeders tracked\n"
-			   << stats.bytes_read << " bytes read\n"
-			   << stats.bytes_written << " bytes written\n";
+		output << "Uptime: " << up_d << " days, "
+				<< (up_h < 10 ? "0" : "") << inttostr(up_h) << ':'
+				<< (up_m < 10 ? "0" : "") << inttostr(up_m) << ':'
+				<< (up_s < 10 ? "0" : "") << inttostr(up_s) << "\n"
+			<< "version: " << version() << "\n"
+			<< stats.opened_connections << " connections opened\n"
+			<< stats.open_connections << " open connections\n"
+			<< stats.connection_rate << " connections/s\n"
+			<< stats.requests << " requests handled\n"
+			<< stats.request_rate << " requests/s\n"
+			<< stats.succ_announcements << " successful announcements\n"
+			<< (stats.announcements - stats.succ_announcements) << " failed announcements\n"
+			<< stats.scrapes << " scrapes\n"
+			<< stats.leechers << " leechers tracked\n"
+			<< stats.seeders << " seeders tracked\n"
+			<< stats.user_queue_size << " items in user queue\n"
+			<< stats.torrent_queue_size << " items in torrent queue\n"
+			<< stats.peer_queue_size << " items in peer queue\n"
+			<< stats.snatch_queue_size << " items in snatch queue\n"
+			<< stats.token_queue_size << " items in token queue\n"
+			<< stats.bytes_read << " bytes read\n"
+			<< stats.bytes_written << " bytes written\n"
+			<< announce_interval << " announce interval\n"
+			<< announce_jitter << " announce jitter\n"
+			;
+	} else if (action == "prom_stats") {
+		time_t uptime = time(NULL) - stats.start_time;
+		output << "ocelot_uptime " << uptime << "\n"
+			<< "ocelot_open_connections " << stats.open_connections << "\n"
+			<< "ocelot_connection_rate " << stats.connection_rate << "\n"
+			<< "ocelot_requests " << stats.requests << "\n"
+			<< "ocelot_request_rate " << stats.request_rate << "\n"
+			<< "ocelot_succ_announcements " << stats.succ_announcements << "\n"
+			<< "ocelot_total_announcements " << stats.announcements << "\n"
+			<< "ocelot_scrapes " << stats.scrapes << "\n"
+			<< "ocelot_leechers " << stats.leechers << "\n"
+			<< "ocelot_seeders " << stats.seeders << "\n"
+			<< "ocelot_user_queue " << stats.user_queue_size << "\n"
+			<< "ocelot_torrent_queue " << stats.torrent_queue_size << "\n"
+			<< "ocelot_peer_queue " << stats.peer_queue_size << "\n"
+			<< "ocelot_snatch_queue " << stats.snatch_queue_size << "\n"
+			<< "ocelot_token_queue " << stats.token_queue_size << "\n"
+			<< "ocelot_bytes_read " << stats.bytes_read << "\n"
+			<< "ocelot_bytes_written " << stats.bytes_written << "\n#";
 	} else if (action == "user") {
 		std::string key = params["key"];
 		if (key.empty()) {
@@ -44,8 +70,14 @@ std::string report(params_type &params, user_list &users_list, client_opts_t &cl
 		} else {
 			user_list::const_iterator u = users_list.find(key);
 			if (u != users_list.end()) {
-				output << u->second->get_leeching() << " leeching" << std::endl
-					   << u->second->get_seeding() << " seeding" << std::endl;
+				output << "{\"id\":" << u->second->get_id()
+					<< ",\"leeching\":" << u->second->get_leeching()
+					<< ",\"seeding\":" << u->second->get_seeding()
+					<< ",\"deleted\":" << u->second->is_deleted()
+					<< ",\"protected\":" << u->second->is_protected()
+					<< ",\"can_leech\":" << u->second->can_leech()
+					<< "}" << "\n";
+				return response(output.str(), client_opts);
 			}
 		}
 	} else {

src/report.h

@@ -1,4 +1,9 @@
-#include <string>
 #include "ocelot.h"
 
-std::string report(params_type &params, user_list &users_list, client_opts_t &client_opts);
+std::string report(
+	params_type &params,
+	user_list &users_list,
+	client_opts_t &client_opts,
+	unsigned int announce_interval,
+	unsigned int announce_jitter 
+);

src/site_comm.cpp

@@ -38,7 +38,7 @@ bool site_comm::all_clear() {
 void site_comm::expire_token(int torrent, int user) {
 	std::stringstream token_pair;
 	token_pair << user << ':' << torrent;
-	if (expire_token_buffer != "") {
+	if (!expire_token_buffer.empty()) {
 		expire_token_buffer += ",";
 	}
 	expire_token_buffer += token_pair.str();
@@ -63,7 +63,7 @@ void site_comm::flush_tokens()
 	if (verbose_flush || qsize > 0) {
 		logger->info("Token expire queue size: " + std::to_string(qsize));
 	}
-	if (expire_token_buffer == "") {
+	if (expire_token_buffer.empty()) {
 		return;
 	}
 	token_queue.push(expire_token_buffer);
@@ -127,6 +127,8 @@ void site_comm::do_flush_tokens()
 				token_queue.pop();
 			} else {
 				logger->error("Response returned with status code " + std::to_string(status_code) + " when trying to expire a token!");
+				// TODO: add exponential backoff
+				std::this_thread::sleep_for(std::chrono::seconds(5));
 			}
 		}
 	} catch (std::exception &er) {

src/worker.cpp

@@ -22,20 +22,21 @@
 
 //---------- Worker - does stuff with input
 worker::worker(config * conf_obj, torrent_list &torrents, user_list &users, std::vector<std::string> &_whitelist, mysql * db_obj, site_comm * sc) :
-	conf(conf_obj), db(db_obj), s_comm(sc), torrents_list(torrents), users_list(users), whitelist(_whitelist), status(OPEN), reaper_active(false)
+	conf(conf_obj), db(db_obj), s_comm(sc), torrents_list(torrents), users_list(users), whitelist(_whitelist), status(OPEN), reaper_active(false), randgen((std::random_device())())
 {
 	logger = spdlog::get("logger");
 	load_config(conf);
 }
 
 void worker::load_config(config * conf) {
-	announce_interval = conf->get_uint("announce_interval");
+	announce_interval   = conf->get_uint("announce_interval");
 	del_reason_lifetime = conf->get_uint("del_reason_lifetime");
-	peers_timeout = conf->get_uint("peers_timeout");
-	numwant_limit = conf->get_uint("numwant_limit");
-	keepalive_enabled = conf->get_uint("keepalive_timeout") != 0;
-	site_password = conf->get_str("site_password");
-	report_password = conf->get_str("report_password");
+	peers_timeout       = conf->get_uint("peers_timeout");
+	numwant_limit       = conf->get_uint("numwant_limit");
+	keepalive_enabled   = conf->get_uint("keepalive_timeout") != 0;
+	site_password       = conf->get_str("site_password");
+	report_password     = conf->get_str("report_password");
+	jitter              = std::uniform_int_distribution<int>(0, conf->get_uint("announce_jitter"));
 }
 
 void worker::reload_config(config * conf) {
@@ -226,7 +227,7 @@ std::string worker::work(const std::string &input, std::string &ip, client_opts_
 	if (action == REPORT) {
 		if (passkey == report_password) {
 			std::lock_guard<std::mutex> ul_lock(db->user_list_mutex);
-			return report(params, users_list, client_opts);
+			return report(params, users_list, client_opts, announce_interval, conf->get_uint("announce_jitter"));
 		} else {
 			return error("Authentication failure", client_opts);
 		}
@@ -585,7 +586,7 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
 
 				// Find out where to begin in the seeder list
 				peer_list::const_iterator i;
-				if (tor.last_selected_seeder == "") {
+				if (tor.last_selected_seeder.empty()) {
 					i = tor.seeders.begin();
 				} else {
 					i = tor.seeders.find(tor.last_selected_seeder);
@@ -712,7 +713,7 @@ std::string worker::announce(const std::string &input, torrent &tor, user_ptr &u
 	output += "e10:incompletei";
 	output += inttostr(tor.leechers.size());
 	output += "e8:intervali";
-	output += inttostr(announce_interval + std::min((size_t)600, tor.seeders.size())); // ensure a more even distribution of announces/second
+	output += inttostr(announce_interval + jitter(randgen));
 	output += "e12:min intervali";
 	output += inttostr(announce_interval);
 	output += "e5:peers";
@@ -983,18 +984,26 @@ std::string worker::update(params_type &params, client_opts_t &client_opts) {
 		conf->set("announce_interval", interval);
 		announce_interval = conf->get_uint("announce_interval");
 		logger->info("Edited announce interval to " + std::to_string(announce_interval));
+	} else if (params["action"] == "update_announce_jitter") {
+		const std::string new_jitter = params["new_announce_jitter"];
+		conf->set("announce_jitter", new_jitter);
+		jitter = std::uniform_int_distribution<int>(0, conf->get_uint("announce_jitter"));
+		logger->info("Edited announce jitter to " + std::to_string(conf->get_uint("announce_jitter")));
 	} else if (params["action"] == "info_torrent") {
+		std::stringstream output;
 		std::string info_hash_hex = params["info_hash"];
 		std::string info_hash = hex_decode(info_hash_hex);
-		logger->info("Info for torrent '" + std::string(info_hash_hex) + "'");
 		std::lock_guard<std::mutex> tl_lock(db->torrent_list_mutex);
 		auto torrent_it = torrents_list.find(info_hash);
+		output << "{\"hash\":" << std::string(info_hash_hex);
 		if (torrent_it != torrents_list.end()) {
-			logger->info("Torrent " + std::to_string(torrent_it->second.id)
-				+ ", freetorrent = " + std::to_string(torrent_it->second.free_torrent));
+			output << ",\"id\":" << std::to_string(torrent_it->second.id)
+				<< ",\"free\":" << std::to_string(torrent_it->second.free_torrent);
 		} else {
-			logger->warn("Failed to find torrent " + info_hash_hex);
+			output << ",\"fail\":1";
 		}
+		output << "}\n";
+		return response(output.str(), client_opts);
 	}
 	return response("success", client_opts);
 }

src/worker.h

@@ -9,6 +9,8 @@
 #include <mutex>
 #include <ctime>
 #include <spdlog/spdlog.h>
+#include <random>
+
 #include "site_comm.h"
 #include "ocelot.h"
 
@@ -27,6 +29,8 @@ class worker {
 		bool reaper_active;
 		time_t cur_time;
 		std::shared_ptr<spdlog::logger> logger;
+		std::mt19937 randgen;
+		std::uniform_int_distribution<int> jitter;
 
 		unsigned int announce_interval;
 		unsigned int del_reason_lifetime;