MXS-1220: Add threads resource

The threads are now a REST API resource exposed via the /maxscale/threads
resource collection.

include/maxscale/worker.h

@@ -15,6 +15,7 @@
 #include <maxscale/cdefs.h>
 #include <maxscale/poll.h>
 #include <maxscale/thread.h>
+#include <maxscale/jansson.h>
 
 MXS_BEGIN_DECLS
 
@@ -142,4 +143,25 @@ MXS_SESSION* mxs_worker_deregister_session(uint64_t id);
  */
 MXS_SESSION* mxs_worker_find_session(uint64_t id);
 
+/**
+ * @brief Convert a worker to JSON format
+ *
+ * @param host Hostname of this server
+ * @param id   ID of the worker
+ *
+ * @return JSON resource representing the worker
+ */
+json_t* mxs_worker_to_json(const char* host, int id);
+
+/**
+ * Convert workers into JSON format
+ *
+ * @param host Hostname of this server
+ *
+ * @return A JSON resource collection of workers
+ *
+ * @see mxs_json_resource()
+ */
+json_t* mxs_worker_list_to_json(const char* host);
+
 MXS_END_DECLS

server/core/maxscale/worker.hh

@@ -159,6 +159,16 @@ public:
      */
     static int64_t get_one_statistic(POLL_STAT what);
 
+    /**
+     * Return this worker's statistics.
+     *
+     * @return Local statistics for this worker.
+     */
+    const STATISTICS& get_local_statistics() const
+    {
+        return m_statistics;
+    }
+
     /**
      * Add a file descriptor to the epoll instance of the worker.
      *

server/core/resource.cc

@@ -28,6 +28,7 @@
 #include "maxscale/service.h"
 #include "maxscale/config_runtime.h"
 #include "maxscale/modules.h"
+#include "maxscale/worker.h"
 
 using std::list;
 using std::string;
@@ -105,6 +106,16 @@ bool Resource::matching_variable_path(const string& path, const string& target)
                 rval = true;
             }
         }
+        else if (path == ":thread")
+        {
+            char* end;
+            int id = strtol(target.c_str(), &end, 10);
+
+            if (*end == '\0' && mxs_worker_get(id))
+            {
+                rval = true;
+            }
+        }
     }
 
     return rval;
@@ -354,21 +365,26 @@ HttpResponse cb_logs(const HttpRequest& request)
 
 HttpResponse cb_flush(const HttpRequest& request)
 {
+    int code = MHD_HTTP_INTERNAL_SERVER_ERROR;
+
     // Flush logs
     if (mxs_log_rotate() == 0)
     {
-        return HttpResponse(MHD_HTTP_NO_CONTENT);
+        code = MHD_HTTP_NO_CONTENT;
-    }
-    else
-    {
-        return HttpResponse(MHD_HTTP_INTERNAL_SERVER_ERROR);
     }
+
+    return HttpResponse(code);
 }
 
-HttpResponse cb_threads(const HttpRequest& request)
+HttpResponse cb_all_threads(const HttpRequest& request)
 {
-    // TODO: Show thread status
+    return HttpResponse(MHD_HTTP_OK, mxs_worker_list_to_json(request.host()));
-    return HttpResponse(MHD_HTTP_OK, mxs_json_resource(request.host(), MXS_JSON_API_THREADS, json_null()));
+}
+
+HttpResponse cb_thread(const HttpRequest& request)
+{
+    int id = atoi(request.last_uri_part().c_str());
+    return HttpResponse(MHD_HTTP_OK, mxs_worker_to_json(request.host(), id));
 }
 
 HttpResponse cb_tasks(const HttpRequest& request)
@@ -425,7 +441,8 @@ public:
         m_get.push_back(SResource(new Resource(cb_get_session, 2, "sessions", ":session")));
 
         m_get.push_back(SResource(new Resource(cb_maxscale, 1, "maxscale")));
-        m_get.push_back(SResource(new Resource(cb_threads, 2, "maxscale", "threads")));
+        m_get.push_back(SResource(new Resource(cb_all_threads, 2, "maxscale", "threads")));
+        m_get.push_back(SResource(new Resource(cb_thread, 3, "maxscale", "threads", ":thread")));
         m_get.push_back(SResource(new Resource(cb_logs, 2, "maxscale", "logs")));
         m_get.push_back(SResource(new Resource(cb_tasks, 2, "maxscale", "tasks")));
         m_get.push_back(SResource(new Resource(cb_all_modules, 2, "maxscale", "modules")));

server/core/worker.cc

@@ -12,11 +12,15 @@
  */
 
 #include "maxscale/worker.hh"
+
 #include <errno.h>
 #include <fcntl.h>
 #include <signal.h>
 #include <stdlib.h>
 #include <unistd.h>
+#include <vector>
+#include <sstream>
+
 #include <maxscale/alloc.h>
 #include <maxscale/atomic.h>
 #include <maxscale/config.h>
@@ -24,13 +28,21 @@
 #include <maxscale/log_manager.h>
 #include <maxscale/platform.h>
 #include <maxscale/semaphore.hh>
+#include <maxscale/json_api.h>
+#include <maxscale/utils.hh>
+
 #include "maxscale/modules.h"
 #include "maxscale/poll.h"
 #include "maxscale/statistics.h"
+#include "maxscale/workertask.hh"
 
 #define WORKER_ABSENT_ID -1
 
 using maxscale::Worker;
+using maxscale::Closer;
+using maxscale::Semaphore;
+using std::vector;
+using std::stringstream;
 
 namespace
 {
@@ -767,6 +779,87 @@ MXS_SESSION* Worker::find_session(uint64_t id)
     return rval;
 }
 
+class WorkerInfoTask: public maxscale::WorkerTask
+{
+public:
+    WorkerInfoTask(const char* host, uint32_t nthreads):
+        m_host(host)
+    {
+        m_data.resize(nthreads);
+    }
+
+    void execute(Worker& worker)
+    {
+        json_t* stats = json_object();
+        const Worker::STATISTICS& s = worker.get_local_statistics();
+        json_object_set_new(stats, "reads", json_integer(s.n_read));
+        json_object_set_new(stats, "writes", json_integer(s.n_write));
+        json_object_set_new(stats, "errors", json_integer(s.n_error));
+        json_object_set_new(stats, "hangups", json_integer(s.n_hup));
+        json_object_set_new(stats, "accepts", json_integer(s.n_accept));
+        json_object_set_new(stats, "blocking_polls", json_integer(s.blockingpolls));
+        json_object_set_new(stats, "event_queue_length", json_integer(s.evq_length));
+        json_object_set_new(stats, "max_event_queue_length", json_integer(s.evq_max));
+        json_object_set_new(stats, "max_exec_time", json_integer(s.maxexectime));
+        json_object_set_new(stats, "max_queue_time", json_integer(s.maxqtime));
+
+        json_t* attr = json_object();
+        json_object_set_new(attr, "stats", stats);
+
+        int idx = worker.get_current_id();
+        stringstream ss;
+        ss << idx;
+
+        json_t* json = json_object();
+        json_object_set_new(json, CN_ID, json_string(ss.str().c_str()));
+        json_object_set_new(json, CN_TYPE, json_string(CN_THREADS));
+        json_object_set_new(json, CN_ATTRIBUTES, attr);
+
+        ss_dassert((size_t)idx < m_data.size());
+        m_data[idx] = json;
+    }
+
+    json_t* resource()
+    {
+        json_t* arr = json_array();
+
+        for (vector<json_t*>::iterator it = m_data.begin(); it != m_data.end(); it++)
+        {
+            json_array_append_new(arr, *it);
+        }
+
+        return mxs_json_resource(m_host, MXS_JSON_API_THREADS, arr);
+    }
+
+    json_t* resource(int id)
+    {
+        return mxs_json_resource(m_host, MXS_JSON_API_THREADS, m_data[id]);
+    }
+
+private:
+    vector<json_t*> m_data;
+    const char*     m_host;
+};
+
+json_t* mxs_worker_to_json(const char* host, int id)
+{
+    Worker* target = Worker::get(id);
+    WorkerInfoTask task(host, id + 1);
+    Semaphore sem;
+
+    target->post(&task, &sem);
+    sem.wait();
+
+    return task.resource(id);
+}
+
+json_t* mxs_worker_list_to_json(const char* host)
+{
+    WorkerInfoTask task(host, config_threadcount());
+    Worker::execute_concurrently(task);
+    return task.resource();
+}
+
 void Worker::run()
 {
     this_thread.current_worker_id = m_id;