exporter: httpserver and collector threads

* Start http server on another thread.
* Expose metrics plain text (string) so that the /perf_counters endpoint
  can retrieve it.
* Introduce stats_period so that the collector updates the metrics every
  'stats_period' seconds.

Signed-off-by: Pere Diaz Bou <pdiazbou@redhat.com>

diff --git a/src/ceph_exporter.cc b/src/ceph_exporter.cc
index 8621ca4e51a2b5d4b2b370e7521718003dec5fe1..37e33bdad62d1d39b459ac04b998d8439ae2432b 100644 (file)
--- a/src/ceph_exporter.cc
+++ b/src/ceph_exporter.cc
@@ -14,6 +14,8 @@
 #include <iostream>
 #include <memory>
 #include <string>
+#include <boost/thread/thread.hpp>
+
 
 
 namespace beast = boost::beast;         // from <boost/beast.hpp>
@@ -21,6 +23,9 @@ namespace http = beast::http;           // from <boost/beast/http.hpp>
 namespace net = boost::asio;            // from <boost/asio.hpp>
 using tcp = boost::asio::ip::tcp;       // from <boost/asio/ip/tcp.hpp>
 
+// FIXME: lets save collector instance here for now.
+DaemonMetricCollector collector;
+
 namespace my_program_state
 {
     std::size_t
@@ -46,9 +51,9 @@ public:
     }
 
     // Initiate the asynchronous operations associated with the connection.
-    void start(std::string& response)
+    void start()
     {
-        read_request(response);
+        read_request();
         check_deadline();
     }
 
@@ -62,7 +67,7 @@ private:
         socket_.get_executor(), std::chrono::seconds(60)};
 
     // Asynchronously receive a complete request message.
-    void read_request(std::string& response)
+    void read_request()
     {
         // auto self = shared_from_this();
 
@@ -75,12 +80,12 @@ private:
             {
                 boost::ignore_unused(bytes_transferred);
                 if(!ec)
-                    process_request(response);
+                    process_request();
             });
     }
 
     // Determine what needs to be done with the request message.
-    void process_request(std::string& response)
+    void process_request()
     {
         response_.version(request_.version());
         response_.keep_alive(false);
@@ -90,7 +95,7 @@ private:
         case http::verb::get:
             response_.result(http::status::ok);
             response_.set(http::field::server, "Beast");
-            create_response(response);
+            create_response();
             break;
 
         default:
@@ -109,12 +114,14 @@ private:
     }
 
     // Construct a response message based on the program state.
-    void create_response(std::string& response)
+    void create_response()
     {
+        std::cout << "Got request on " << request_.target() << std::endl;
         if(request_.target() == "/perf_counters")
         {
             response_.set(http::field::content_type, "text/plain");
-            beast::ostream(response_.body()) << "Perf Counters\n" << response << std::endl;
+            std::string metrics = collector.get_metrics();
+            beast::ostream(response_.body()) << "Perf Counters\n" << metrics << std::endl;
         }
         else
         {
@@ -159,14 +166,15 @@ private:
 };
 
 // "Loop" forever accepting new connections.
-void http_server(tcp::acceptor& acceptor, tcp::socket& socket, std::string& response)
+void http_server(tcp::acceptor& acceptor, tcp::socket& socket)
 {
   acceptor.async_accept(socket,
       [&](beast::error_code ec)
       {
+          std::cout << "async accept" << std::endl;
           if(!ec)
-              std::make_shared<http_connection>(std::move(socket))->start(response);
-          http_server(acceptor, socket, response);
+              std::make_shared<http_connection>(std::move(socket))->start();
+          http_server(acceptor, socket);
       });
 }
 
@@ -181,20 +189,11 @@ std::string dns_lookup(std::string hostname) {
     return ip_address;
 }
 
-int main(int argc, char** argv) {
-  // TODO: daemonize
-  std::cout << "inside exporter" << std::endl;
-
-  DaemonMetricCollector collector;
-  int stats_period = 5;
-  std::string response;
-  response = collector.main();
-  std::cout << "response: " << std::endl;
-
+void http_server_thread_entrypoint() {
   try
     {
         std::string hostname = ceph_get_short_hostname();
-        
+
         std::string ip_address = dns_lookup(hostname);
         auto const address = net::ip::make_address(ip_address);
         unsigned short port = static_cast<unsigned short>(std::atoi("9085"));
@@ -203,13 +202,24 @@ int main(int argc, char** argv) {
 
         tcp::acceptor acceptor{ioc, {address, port}};
         tcp::socket socket{ioc};
-        http_server(acceptor, socket, response);
-
+        http_server(acceptor, socket);
+        std::cout << "Http server started" << std::endl;
         ioc.run();
     }
     catch(std::exception const& e)
     {
         std::cerr << "Error: " << e.what() << std::endl;
-        return EXIT_FAILURE;
+        exit(EXIT_FAILURE);
     }
 }
+
+int main(int argc, char** argv) {
+  // TODO: daemonize
+  std::cout << "inside exporter" << std::endl;
+
+  std::cout << "Starting http server thread..." << std::endl;
+  boost::thread server_thread(http_server_thread_entrypoint);
+  std::cout << "Starting collector..." << std::endl;
+  collector.main();
+  server_thread.join();
+}

diff --git a/src/mgr/DaemonMetricCollector.cc b/src/mgr/DaemonMetricCollector.cc
index 173333b9439b4b9e26b7fcba3de158f8e246a15c..9a74359c71681e3e4b37d6c2fc605cefae9ea619 100644 (file)
--- a/src/mgr/DaemonMetricCollector.cc
+++ b/src/mgr/DaemonMetricCollector.cc
@@ -13,12 +13,27 @@ using json_object = boost::json::object;
 using json_value = boost::json::value;
 using json_array = boost::json::array;
 
-std::string DaemonMetricCollector::main() {
-  update_sockets();
-  send_requests();
-  // result.push_back(reqs);
+void DaemonMetricCollector::request_loop(boost::asio::deadline_timer &timer) {
+  timer.async_wait([&](const boost::system::error_code& e) {
+    std::cerr << e << std::endl;
+    update_sockets();
+    std::cout << "updating metrics" << std::endl;
+    send_requests();
+    timer.expires_from_now(boost::posix_time::seconds(stats_period));
+    request_loop(timer);
+  });
+}
+void DaemonMetricCollector::main() {
+  stats_period = 5; // TODO: let's do 5 for now and expose this to change in the future
+  boost::asio::io_service io;
+  boost::asio::deadline_timer timer(io, boost::posix_time::seconds(stats_period));
+  std::cout << "before request" << std::endl;
+  request_loop(timer);
+  io.run();
+}
 
-  return result;
+std::string DaemonMetricCollector::get_metrics() {
+  return metrics;
 }
 
 template <class T>
@@ -46,7 +61,8 @@ std::string boost_string_to_std(boost::json::string js) {
   return res;
 }
 
-std::string DaemonMetricCollector::send_requests() {
+void DaemonMetricCollector::send_requests() {
+  std::string result;
   for(auto client : clients) {
     AdminSocketClient &sock_client = client.second;
     std::string daemon_name = client.first;
@@ -117,11 +133,11 @@ std::string DaemonMetricCollector::send_requests() {
       }
     }
   }
-  return result;
+  metrics = result;
 }
 
 void DaemonMetricCollector::update_sockets() {
-  std::string path = "/tmp/ceph-asok.Qcdpny/";
+  std::string path = "/tmp/ceph-asok.7MK7oH/";
   for (const auto & entry : std::filesystem::directory_iterator(path)) {
     std::string daemon_socket_name = entry.path().filename().string();
     // remove .asok

diff --git a/src/mgr/DaemonMetricCollector.h b/src/mgr/DaemonMetricCollector.h
index ce4bf63f5343d88a5ad48a508eef2fa95dea1a6d..a9b6c81b492d1818dd7bcc2dced973b578ecb678 100644 (file)
--- a/src/mgr/DaemonMetricCollector.h
+++ b/src/mgr/DaemonMetricCollector.h
@@ -8,19 +8,23 @@
 #include <string>
 #include <map>
 #include <vector>
+#include <boost/asio.hpp>
 
 class DaemonMetricCollector {
  public:
   int i;
-  std::string main();
+  void main();
+  std::string get_metrics();
 
 private:
   // TODO: add clients
   //       check removed sockets
   //       list dir of sockets
   std::map<std::string, AdminSocketClient> clients;
-  std::string result;
+  std::string metrics;
+  int stats_period; // time to wait before sending requests again
   void update_sockets();
-  std::string send_requests();
+  void request_loop(boost::asio::deadline_timer &timer);
+  void send_requests();
   void start_mgr_connection();
 };