added (untested) statsd for the cmake network.

5 files changed, 144 insertions, 0 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index c01a44e..bdede1a 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -54,5 +54,6 @@ add_subdirectory(libcmix-network)
 add_subdirectory(libcmix-common)
 add_subdirectory(node)
 add_subdirectory(client)
+add_subdirectory(statsd)
 
 add_subdirectory(scratchpad)
diff --git a/statsd/CMakeLists.txt b/statsd/CMakeLists.txt
new file mode 100644
index 0000000..73dbf68
--- /dev/null
+++ b/statsd/CMakeLists.txt
@@ -0,0 +1,16 @@
+find_package(Boost COMPONENTS system program_options REQUIRED)
+
+add_executable(statsd
+	main.cpp
+	stats.hpp stats.cpp
+)
+
+target_link_libraries(statsd
+	PRIVATE Boost::boost
+	PRIVATE Boost::program_options
+	PRIVATE Boost::system
+	PRIVATE cmix-protobuf
+	PRIVATE cmix-network
+	PRIVATE cmix-common
+	PRIVATE log
+)
\ No newline at end of file
diff --git a/statsd/main.cpp b/statsd/main.cpp
new file mode 100644
index 0000000..436e8ad
--- /dev/null
+++ b/statsd/main.cpp
@@ -0,0 +1,45 @@
+
+#include "stats.hpp"
+
+#include "logging.hpp"
+
+#include <boost/program_options.hpp>
+
+#include <iostream>
+
+int main(int argc, char* argv[]) {
+	namespace po = boost::program_options;
+	
+	init_logging(boost::log::trivial::severity_level::trace, "statsd");
+	
+	po::options_description desc("Allowed options");
+	desc.add_options()
+		("help,h", "produce help message.")
+		("port,p", po::value<uint16_t>()->default_value(9200), "Set listening port.")
+		("enable_v4", po::value<bool>()->default_value(true), "Enable/disable ipv4 accept support.")
+		("interface4,4", po::value<std::string>()->default_value("0.0.0.0"), "Set the ipv4 address to listen on.")
+		("enable_v6", po::value<bool>()->default_value(true), "Enable/disable ipv6 accept support.")
+		("interface6,6", po::value<std::string>()->default_value("::"), "Set the ipv6 address to listen on.")
+	;
+
+	po::variables_map vm;
+	po::store(po::parse_command_line(argc, argv, desc), vm);
+	po::notify(vm);
+	
+	if (vm.count("help")) {
+		std::cout << desc << "\n";
+		return 0;
+	}
+	
+	bool en4 = vm["enable_v4"].as<bool>();
+	std::string if4 = vm["interface4"].as<std::string>();
+	bool en6 = vm["enable_v6"].as<bool>();
+	std::string if6 = vm["interface6"].as<std::string>();
+	uint16_t port = vm["port"].as<uint16_t>();
+	
+	ListenSettings lsettings{en4, if4, en6, if6, port, false, "", "", ""};
+	Stats stats(lsettings);
+	
+	stats.run();
+	
+}
diff --git a/statsd/stats.cpp b/statsd/stats.cpp
new file mode 100644
index 0000000..9eeb140
--- /dev/null
+++ b/statsd/stats.cpp
@@ -0,0 +1,49 @@
+#include "stats.hpp"
+
+void Stats::accept_connection(std::unique_ptr<boost::asio::ip::tcp::socket>&& socket) {
+	auto it = connections.emplace(connections.end(), std::move(socket));
+	it->on_done([this, it](){
+		connections.erase(it);
+	});
+	it->async_receive([it, this](cmix_proto::CMixMessage const& message) {
+		handle_message(it, message);
+	});
+}
+
+void Stats::handle_performance(std::list<Receiver>::iterator it, const cmix_proto::Performance& perf) {
+	data[perf.node()][perf.column() + "wall_time"].push_back(std::stol(perf.wall_time()));
+	data[perf.node()][perf.column() + "system_time"].push_back(std::stol(perf.system_time()));
+	data[perf.node()][perf.column() + "user_time"].push_back(std::stol(perf.user_time()));
+	
+	it->async_receive([it, this](cmix_proto::CMixMessage const& message) {
+		handle_message(it, message);
+	});
+}
+
+void Stats::handle_message(std::list<Receiver>::iterator it, cmix_proto::CMixMessage message) {
+	switch(message.contents_case()) {
+	case cmix_proto::CMixMessage::ContentsCase::kPerformance: {
+		BOOST_LOG_TRIVIAL(trace) << "Handling performance";
+		handle_performance(it, message.performance());
+		break;
+	}
+	case cmix_proto::CMixMessage::ContentsCase::kBye: {
+		BOOST_LOG_TRIVIAL(trace) << "Handling Bye";
+		
+		break;
+	}
+	default: {
+		BOOST_LOG_TRIVIAL(error) << "handle_message: CMixMessage contains unknown contents.";
+		connections.erase(it);
+	}
+	}
+}
+
+Stats::Stats(ListenSettings lsettings)
+: io_service()
+, server(io_service, lsettings, [this](std::unique_ptr<boost::asio::ip::tcp::socket>&& socket){accept_connection(std::move(socket));})
+{}
+
+void Stats::run() {
+	io_service.run();
+}
diff --git a/statsd/stats.hpp b/statsd/stats.hpp
new file mode 100644
index 0000000..673022e
--- /dev/null
+++ b/statsd/stats.hpp
@@ -0,0 +1,33 @@
+#pragma once
+
+#include "server.hpp"
+#include "receiver.hpp"
+
+#include "cmix.pb.h"
+
+#include <boost/asio/io_service.hpp>
+
+#include <list>
+
+class Stats {
+	boost::asio::io_service io_service;
+	Server server;
+	std::list<Receiver> connections;
+	
+	typedef std::vector<uint64_t> Column;
+	typedef std::map<std::string, Column> Table;
+	typedef std::map<std::string, Table> Tables;
+	
+	Tables data;
+
+	void accept_connection(std::unique_ptr<boost::asio::ip::tcp::socket>&& socket);
+	
+	void handle_performance(std::list<Receiver>::iterator it, cmix_proto::Performance const& perf);
+	
+	void handle_message(std::list<Receiver>::iterator it, cmix_proto::CMixMessage message);
+
+public:
+	Stats(ListenSettings lsettings);
+	
+	void run();
+};
\ No newline at end of file