Implement vitastor-cli serve command to serve simple HTTP API

src/cmd/cli_serve.cpp

@@ -0,0 +1,428 @@
+// Copyright (c) Vitaliy Filippov, 2019+
+// License: VNPL-1.1 (see README.md for details)
+
+#include <signal.h>
+#include <sys/socket.h>
+#include <netinet/tcp.h>
+#include <stdexcept>
+#include "cli.h"
+#include "cluster_client.h"
+#include "epoll_manager.h"
+#include "http_client.h"
+#include "str_util.h"
+#include "json_util.h"
+#include "addr_util.h"
+
+struct cli_serve_conn_t
+{
+    int peer_fd = 0;
+    std::string peer_addr;
+    http_co_t *co = NULL;
+    cli_tool_t *p = NULL;
+    cli_result_t result;
+    bool keepalive = false;
+    bool closed = false;
+    timespec request_time;
+    std::string request_method;
+    std::string request_path;
+    std::string request_body;
+    std::function<bool(cli_result_t &)> action_cb;
+};
+
+struct cli_serve_path_t
+{
+    std::string cmd;
+    bool allow_get;
+};
+
+// Serve vitastor-cli commands over HTTP in JSON format
+struct cli_serve_t
+{
+    std::map<std::string, cli_serve_path_t> cmd_paths = {
+        {"data/delete",   {"rm-data", false}},
+        {"data/describe", {"describe", true}},
+        {"data/fix",      {"fix", false}},
+        {"data/merge",    {"merge-data", false}},
+        {"image/create",  {"create", false}},
+        {"image/delete",  {"rm", false}},
+        {"image/flatten", {"flatten", false}},
+        {"image/list",    {"ls", true}},
+        {"image/modify",  {"modify", false}},
+        {"osd/alloc",     {"alloc-osd", false}},
+        {"osd/delete",    {"rm-osd", false}},
+        {"osd/list",      {"ls-osd", true}},
+        {"osd/modify",    {"modify-osd", false}},
+        {"pg/list",       {"ls-pgs", true}},
+        {"pool/create",   {"create-pool", false}},
+        {"pool/delete",   {"rm-pool", false}},
+        {"pool/list",     {"pools", true}},
+        {"pool/modify",   {"modify-pool", false}},
+        {"status",        {"status", true}},
+    };
+
+    cli_tool_t *parent = NULL;
+    json11::Json options;
+    cli_result_t result;
+
+    bool log_body = false;
+    bool stop = false;
+    std::vector<std::string> bind_addresses;
+    int port = 0;
+    int listen_backlog = 0;
+    bool ssl = false;
+    std::vector<int> listen_fds;
+    http_context_t *http_ctx = NULL;
+    std::set<cli_serve_conn_t*> connections;
+
+    int state = 0;
+
+    bool is_done()
+    {
+        return state == 100;
+    }
+
+    void loop()
+    {
+        if (state == 1)
+            goto resume_1;
+        else if (state == 2)
+            goto resume_2;
+        else if (state == 100)
+            return;
+        if (options["bind_address"].is_string())
+            bind_addresses = explode(" ", options["bind_address"].string_value(), true);
+        else
+            bind_addresses.push_back("127.0.0.1");
+        port = options["port"].uint64_value();
+        if (!port)
+            port = 8080;
+        else if (port < 0 || port > 65535)
+        {
+            result = (cli_result_t){ .err = EINVAL, .text = "HTTP port can't be larger than 65536" };
+            state = 100;
+            return;
+        }
+        listen_backlog = options["listen_backlog"].uint64_value();
+        if (!listen_backlog)
+            listen_backlog = 128;
+        ssl = json_is_true(options["ssl"]);
+        if (ssl)
+        {
+            std::string ssl_cert = options["ssl_cert"].string_value();
+            std::string ssl_key = options["ssl_key"].string_value();
+            std::string ssl_ca = options["ssl_ca"].string_value();
+            std::string error;
+            http_ctx = http_context_init(ssl_cert, ssl_key, ssl_ca, ssl_ca != "", error);
+            if (error != "")
+            {
+                result = (cli_result_t){ .err = EINVAL, .text = error };
+                state = 100;
+                return;
+            }
+        }
+        for (auto & bind_address: bind_addresses)
+        {
+            int listen_fd = create_and_bind_socket(bind_address, port, listen_backlog, NULL);
+            fcntl(listen_fd, F_SETFL, fcntl(listen_fd, F_GETFL, 0) | O_NONBLOCK);
+            parent->epmgr->set_fd_handler(listen_fd, false, [this](int fd, int events)
+            {
+                accept_connections(fd);
+            });
+            listen_fds.push_back(listen_fd);
+        }
+    resume_1:
+        if (!stop)
+        {
+            state = 1;
+            return;
+        }
+        for (auto conn: connections)
+        {
+            http_close(conn->co);
+        }
+    resume_2:
+        // Wait for all connections to finish
+        if (connections.size() > 0)
+        {
+            state = 2;
+            return;
+        }
+        if (http_ctx)
+        {
+            http_context_destroy(http_ctx);
+            http_ctx = NULL;
+        }
+        state = 100;
+    }
+
+    void accept_connections(int listen_fd)
+    {
+        sockaddr_storage addr;
+        socklen_t peer_addr_size = sizeof(addr);
+        int peer_fd;
+        while ((peer_fd = accept(listen_fd, (sockaddr*)&addr, &peer_addr_size)) >= 0)
+        {
+            auto peer_addr_str = addr_to_string(addr);
+            assert(peer_fd != 0);
+            timespec ts;
+            clock_gettime(CLOCK_REALTIME, &ts);
+            printf("[%s.%03ju] New connection %d from %s\n", format_datetime(ts.tv_sec).c_str(), (uint64_t)ts.tv_nsec/1000000,
+                peer_fd, peer_addr_str.c_str());
+            fcntl(peer_fd, F_SETFL, fcntl(peer_fd, F_GETFL, 0) | O_NONBLOCK);
+            int one = 1;
+            setsockopt(peer_fd, SOL_TCP, TCP_NODELAY, &one, sizeof(one));
+            cli_serve_conn_t *conn = new cli_serve_conn_t;
+            conn->peer_fd = peer_fd;
+            conn->peer_addr = peer_addr_str;
+            conn->co = http_init(parent->epmgr->tfd, http_ctx);
+            http_serve(conn->co, peer_fd, (http_options_t){ .ssl = ssl }, [this, conn](http_message_t *msg)
+            {
+                process_request(conn, msg);
+            });
+            connections.insert(conn);
+            // Try to accept next connection
+            peer_addr_size = sizeof(addr);
+        }
+        if (peer_fd == -1 && errno != EAGAIN)
+        {
+            throw std::runtime_error(std::string("accept: ") + strerror(errno));
+        }
+    }
+
+    int map_to_http(int err, std::string *text)
+    {
+        int code = 0;
+        if (err == EINVAL)
+        {
+            code = 400;
+            if (text)
+                *text = "Bad Request";
+        }
+        else if (err == EOPNOTSUPP)
+        {
+            code = 404;
+            if (text)
+                *text = "Not Found";
+        }
+        else if (err == ENOSYS)
+        {
+            code = 405;
+            if (text)
+                *text = "Method Not Allowed";
+        }
+        else if (err == EAGAIN)
+        {
+            code = 409;
+            if (text)
+                *text = "Update Conflict";
+        }
+        else if (err == ENOTEMPTY || err == EEXIST || err == ENOENT || err == EBUSY)
+        {
+            code = 412;
+            if (text)
+                *text = "Precondition Failed";
+        }
+        else /*if (err == EIO || err == EBADF)*/
+        {
+            code = 500;
+            if (text)
+                *text = "Internal Server Error";
+        }
+        return code;
+    }
+
+    std::string cli_http_response(cli_serve_conn_t *conn)
+    {
+        timespec now;
+        clock_gettime(CLOCK_REALTIME, &now);
+        int code = 200;
+        std::string response;
+        if (conn->result.err)
+        {
+            std::string status_line;
+            code = map_to_http(conn->result.err, &status_line);
+            response = "HTTP/1.1 "+std::to_string(code)+" "+status_line+"\r\n";
+        }
+        else
+        {
+            response = "HTTP/1.1 200 OK\r\n";
+        }
+        response += (conn->keepalive
+            ? "Connection: keep-alive\r\n"
+            : "Connection: close\r\n");
+        std::string body;
+        if (!conn->result.data.is_null())
+        {
+            response += "Content-Type: application/json\r\n";
+            body = conn->result.data.dump();
+        }
+        else
+        {
+            response += "Content-Type: text/plain; charset=utf-8\r\n";
+            body = conn->result.text;
+        }
+        response += "Content-Length: "+std::to_string(body.size())+"\r\n\r\n";
+        response += body;
+        if (conn->request_method.find("\n") != std::string::npos)
+            conn->request_method = str_replace(conn->request_method, "\n", "%0a");
+        if (conn->request_method.find(" ") != std::string::npos)
+            conn->request_method = str_replace(conn->request_method, " ", "%20");
+        if (conn->request_path.find("\n") != std::string::npos)
+            conn->request_path = str_replace(conn->request_path, "\n", "%0a");
+        if (conn->request_path.find(" ") != std::string::npos)
+            conn->request_path = str_replace(conn->request_path, " ", "%20");
+        uint64_t response_time = (now.tv_sec-conn->request_time.tv_sec)*1000 + (now.tv_nsec-conn->request_time.tv_nsec)/1000000;
+        printf("[%s.%03ju] %s %s %s %d %.03f sec\n",
+            format_datetime(now.tv_sec).c_str(), (uint64_t)now.tv_nsec/1000000,
+            conn->peer_addr.c_str(), conn->request_method.c_str(), conn->request_path.c_str(), code,
+            response_time/1000.0);
+        if (log_body)
+        {
+            if (conn->request_body.find("\n") != std::string::npos)
+                conn->request_body = str_replace(conn->request_body, "\n", " ");
+            if (conn->request_body.size())
+                printf("  %s\n", conn->request_body.c_str());
+            printf("  %s\n", body.c_str());
+        }
+        return response;
+    }
+
+    void process_request(cli_serve_conn_t *conn, http_message_t *msg)
+    {
+        timespec ts;
+        clock_gettime(CLOCK_REALTIME, &ts);
+        if (!msg->error.empty())
+        {
+            // connection is closed
+            fprintf(stderr, "[%s.%03ju] Connection %d closed: %s\n", format_datetime(ts.tv_sec).c_str(),
+                (uint64_t)ts.tv_nsec/1000000, conn->peer_fd, msg->error.c_str());
+            if (conn->p)
+                conn->closed = true;
+            else
+            {
+                connections.erase(conn);
+                http_destroy(conn->co);
+                delete conn;
+            }
+            return;
+        }
+        conn->keepalive = msg->headers.find("connection") != msg->headers.end() &&
+            msg->headers.at("connection") == "keep-alive";
+        conn->p = new cli_tool_t;
+        conn->p->iodepth = parent->iodepth;
+        conn->p->parallel_osds = parent->parallel_osds;
+        conn->p->json_output = true;
+        conn->p->ringloop = parent->ringloop;
+        conn->p->epmgr = parent->epmgr;
+        conn->p->cli = parent->cli;
+        conn->p->is_command_line = false;
+        // Parse request
+        auto req_line = explode(" ", msg->status_line, true);
+        if (req_line.size() < 2)
+        {
+            if (req_line[0] == "")
+                req_line[0] = "-";
+            req_line.push_back("-");
+        }
+        conn->request_time = ts;
+        conn->request_method = std::move(req_line[0]);
+        conn->request_path = std::move(req_line[1]);
+        conn->request_body = std::move(msg->body);
+        auto ctype = msg->headers["content-type"];
+        if (conn->request_method != "GET" && conn->request_method != "POST")
+        {
+            conn->result = { .err = ENOSYS, .text = "Unsupported request method "+conn->request_method };
+        }
+        else if (ctype != (conn->request_method == "GET" ? "" : "application/json"))
+        {
+            conn->result = { .err = EINVAL, .text = "Unsupported Content-Type: "+ctype+" for "+conn->request_method+" requests" };
+        }
+        else
+        {
+            auto uri = explode("?", conn->request_path, true);
+            uri[0] = trim(uri[0], "/");
+            auto cmd_it = cmd_paths.find(uri[0]);
+            if (uri[0] == "")
+            {
+                std::string text = "Supported APIs:\n\n";
+                for (auto & pp: cmd_paths)
+                {
+                    text += (pp.second.allow_get ? "- GET" : "- POST") + (" /" + pp.first) + "\n";
+                }
+                conn->result = { .text = text };
+            }
+            else if (cmd_it == cmd_paths.end())
+            {
+                conn->result = { .err = EOPNOTSUPP, .text = "unknown command: "+uri[0] };
+            }
+            else if (conn->request_method == "GET" && !cmd_it->second.allow_get)
+            {
+                conn->result = { .err = ENOSYS, .text = "method /"+uri[0]+" only allows POST requests" };
+            }
+            else
+            {
+                std::string error;
+                json11::Json::object cfg;
+                if (conn->request_method == "POST")
+                {
+                    cfg = json11::Json::parse(conn->request_body, error).object_items();
+                }
+                else
+                {
+                    // Parse URI
+                    cfg = parse_uri_params(uri[1]);
+                }
+                if (error != "")
+                {
+                    conn->result = { .err = EINVAL, .text = "Invalid JSON in body: "+error };
+                }
+                else
+                {
+                    cfg["command"] = json11::Json::array{cmd_it->second.cmd};
+                    conn->p->parse_api_opts(cfg);
+                    conn->action_cb = conn->p->start(cfg, conn->result);
+                }
+            }
+        }
+        if (!conn->action_cb)
+        {
+            http_reply(conn->co, cli_http_response(conn));
+            delete conn->p;
+            conn->p = NULL;
+            return;
+        }
+        conn->p->loop_and_wait(conn->action_cb, [this, conn](const cli_result_t & r)
+        {
+            conn->result = r;
+            conn->action_cb = NULL;
+            delete conn->p;
+            conn->p = NULL;
+            if (!conn->closed)
+                http_reply(conn->co, cli_http_response(conn));
+            else
+            {
+                connections.erase(conn);
+                http_destroy(conn->co);
+                delete conn;
+            }
+        });
+    }
+};
+
+std::function<bool(cli_result_t &)> cli_tool_t::start_serve(json11::Json cfg)
+{
+    auto server = new cli_serve_t();
+    server->parent = this;
+    server->options = cfg;
+    return [server](cli_result_t & result)
+    {
+        server->loop();
+        if (server->is_done())
+        {
+            result = server->result;
+            delete server;
+            return true;
+        }
+        return false;
+    };
+}