http: embed http server in both meta and replica

include/dsn/dist/replication/meta_service_app.h

@@ -50,6 +50,9 @@ class test_checker;
 }
 
 namespace dsn {
+
+class http_server;
+
 namespace service {
 
 class meta_service_app : public service_app
@@ -69,6 +72,7 @@ class meta_service_app : public service_app
     friend class ::dsn::replication::replication_checker;
     friend class ::dsn::replication::test::test_checker;
     std::unique_ptr<dsn::replication::meta_service> _service;
+    std::unique_ptr<http_server> _http_server;
 };
 }
 }

include/dsn/dist/replication/replication_service_app.h

@@ -39,6 +39,9 @@
 #include <dsn/cpp/service_app.h>
 
 namespace dsn {
+
+class http_server;
+
 namespace replication {
 
 class replication_checker;
@@ -68,6 +71,7 @@ class replication_service_app : public ::dsn::service_app
     friend class ::dsn::replication::replication_checker;
     friend class ::dsn::replication::test::test_checker;
     replica_stub_ptr _stub;
+    std::unique_ptr<http_server> _http_server;
 
     static const char *replica_service_app_info(int argc, char **argv);
     static void replica_service_app_info_free(const char *response);

include/dsn/tool-api/http_server.h

@@ -0,0 +1,88 @@
+// Copyright (c) 2018, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#pragma once
+
+#include <dsn/c/api_common.h>
+#include <dsn/tool-api/rpc_message.h>
+#include <dsn/cpp/serverlet.h>
+#include <dsn/utility/errors.h>
+
+namespace dsn {
+
+struct http_request
+{
+    static error_with<http_request> parse(dsn_message_t msg);
+
+    std::pair<std::string, std::string> service_method;
+    blob body;
+    blob full_url;
+};
+
+enum class http_status_code
+{
+    ok,                    // 200
+    bad_request,           // 400
+    not_found,             // 404
+    internal_server_error, // 500
+};
+
+extern std::string http_status_code_to_string(http_status_code code);
+
+struct http_response
+{
+    std::string body;
+    http_status_code status_code{http_status_code::ok};
+    std::string content_type = "text/plain";
+
+    ref_ptr<message_ex> to_message(dsn_message_t req) const;
+};
+
+class http_service
+{
+public:
+    typedef std::function<void(const http_request &req, http_response &resp)> http_callback;
+
+    virtual std::string path() const = 0;
+
+    void register_handler(std::string path, http_callback cb)
+    {
+        _cb_map.emplace(std::move(path), std::move(cb));
+    }
+
+    void call(const http_request &req, http_response &resp)
+    {
+        auto it = _cb_map.find(req.service_method.second);
+        if (it != _cb_map.end()) {
+            it->second(req, resp);
+        } else {
+            resp.status_code = http_status_code::bad_request;
+            resp.body = "method not found";
+        }
+    }
+
+private:
+    std::map<std::string, http_callback> _cb_map;
+};
+
+class http_server : public serverlet<http_server>
+{
+public:
+    http_server();
+
+    ~http_server() override = default;
+
+    void add_service(http_service *service);
+
+    void serve(dsn_message_t msg);
+
+private:
+    std::map<std::string, std::unique_ptr<http_service>> _service_map;
+};
+
+/// The rpc code for all the HTTP RPCs.
+/// Since http is used only for system monitoring, it is restricted to lowest priority.
+DEFINE_TASK_CODE_RPC(RPC_HTTP_SERVICE, TASK_PRIORITY_LOW, THREAD_POOL_DEFAULT);
+
+} // namespace dsn

include/dsn/tool-api/rpc_message.h

@@ -67,7 +67,11 @@ struct fast_code
 
 typedef struct message_header
 {
+    // For thrift protocol this is "THFT".
+    // For dsn protocol this is "RDSN".
+    // For http protocol this is either a "GET " or "POST".
     uint32_t hdr_type;
+
     uint32_t hdr_version;
     uint32_t hdr_length;
     uint32_t hdr_crc32;
@@ -154,8 +158,19 @@ class message_ex : public ref_counter,
                                               int thread_hash = 0,
                                               uint64_t partition_hash = 0);
 
+    /// This method is only used for receiving request.
+    /// The returned message:
+    ///   - msg->buffers[0] = message_header
+    ///   - msg->buffers[1] = data
     DSN_API static message_ex *create_receive_message_with_standalone_header(const blob &data);
+
+    /// The returned message:
+    ///   - msg->buffers[0] = message_header
+    ///   - msg->_is_read = false
+    ///   - msg->_rw_index = 0
+    ///   - msg->_rw_offset = 48 (size of message_header)
     DSN_API message_ex *create_response();
+
     DSN_API message_ex *copy(bool clone_content, bool copy_for_receive);
     DSN_API message_ex *copy_and_prepare_send(bool clone_content);
 
@@ -164,7 +179,6 @@ class message_ex : public ref_counter,
     //
     DSN_API void write_next(void **ptr, size_t *size, size_t min_size);
     DSN_API void write_commit(size_t size);
-    DSN_API void write_append(const blob &data);
     DSN_API bool read_next(void **ptr, size_t *size);
     bool read_next(blob &data);
     DSN_API void read_commit(size_t size);

src/core/CMakeLists.txt

@@ -11,4 +11,5 @@ add_library(dsn_runtime STATIC
         $<TARGET_OBJECTS:dsn.tools.common>
         $<TARGET_OBJECTS:dsn.tools.hpc>
         $<TARGET_OBJECTS:dsn.tools.simulator>
+        $<TARGET_OBJECTS:dsn.tools.http>
         )

src/core/core/rpc_message.cpp

@@ -558,22 +558,6 @@ void message_ex::write_commit(size_t size)
     this->header->body_length += (int)size;
 }
 
-void message_ex::write_append(const blob &data)
-{
-    // printf("%p %s\n", this, __FUNCTION__);
-    dassert(!this->_is_read && this->_rw_committed,
-            "there are pending msg write not committed"
-            ", please invoke dsn_msg_write_next and dsn_msg_write_commit in pairs");
-
-    int size = data.length();
-    if (size > 0) {
-        this->_rw_index++;
-        this->_rw_offset += size;
-        this->buffers.push_back(data);
-        this->header->body_length += size;
-    }
-}
-
 bool message_ex::read_next(void **ptr, size_t *size)
 {
     // printf("%p %s %d\n", this, __FUNCTION__, utils::get_current_tid());

src/core/tests/fail_point_test.cpp

@@ -49,18 +49,18 @@ TEST(fail_point, print)
 TEST(fail_point, frequency_and_count)
 {
     fail_point p;
-    p.set_action("80%100*return()");
+    p.set_action("80%10000*return()");
 
     int cnt = 0;
     double times = 0;
-    while (cnt < 100) {
+    while (cnt < 10000) {
         if (p.eval() != nullptr) {
             cnt++;
         }
         times++;
     }
-    ASSERT_TRUE(100 / 0.9 < times);
-    ASSERT_TRUE(100 / 0.7 > times);
+    ASSERT_TRUE(10000 / 0.9 < times);
+    ASSERT_TRUE(10000 / 0.7 > times);
 
     for (int i = 0; i < times; i++) {
         ASSERT_EQ(p.eval(), nullptr);

src/core/tests/http_server_test.cpp

@@ -0,0 +1,43 @@
+// Copyright (c) 2018, Xiaomi, Inc.  All rights reserved.
+// This source code is licensed under the Apache License Version 2.0, which
+// can be found in the LICENSE file in the root directory of this source tree.
+
+#include <dsn/tool-api/http_server.h>
+#include <gtest/gtest.h>
+
+namespace dsn {
+
+TEST(http_server, parse_url)
+{
+    struct test_case
+    {
+        std::string url;
+
+        error_code err;
+        std::pair<std::string, std::string> result;
+    } tests[] = {
+        {"http://127.0.0.1:34601", ERR_OK, {"", ""}},
+        {"http://127.0.0.1:34601/", ERR_OK, {"", ""}},
+        {"http://127.0.0.1:34601///", ERR_OK, {"", ""}},
+        {"http://127.0.0.1:34601/threads", ERR_OK, {"threads", ""}},
+        {"http://127.0.0.1:34601/threads/", ERR_OK, {"threads", ""}},
+        {"http://127.0.0.1:34601//pprof/heap/", ERR_OK, {"pprof", "heap"}},
+        {"http://127.0.0.1:34601//pprof///heap", ERR_OK, {"pprof", "heap"}},
+        {"http://127.0.0.1:34601/pprof/heap/arg/", ERR_INVALID_PARAMETERS, {}},
+    };
+
+    for (auto tt : tests) {
+        ref_ptr<message_ex> m = message_ex::create_receive_message_with_standalone_header(
+            blob::create_from_bytes(std::string("POST")));
+        m->buffers.emplace_back(blob::create_from_bytes(std::string(tt.url)));
+
+        auto res = http_request::parse(m.get());
+        if (res.is_ok()) {
+            ASSERT_EQ(res.get_value().service_method, tt.result) << tt.url;
+        } else {
+            ASSERT_EQ(res.get_error().code(), tt.err);
+        }
+    }
+}
+
+} // namespace dsn

src/core/tools/CMakeLists.txt

@@ -1,3 +1,4 @@
 add_subdirectory(common)
 add_subdirectory(simulator)
 add_subdirectory(hpc)
+add_subdirectory(http)