xref: /unit/src/nxt_controller.c (revision 1526)

/*
 * Copyright (C) Igor Sysoev
 * Copyright (C) Valentin V. Bartenev
 * Copyright (C) NGINX, Inc.
 */

#include <nxt_main.h>
#include <nxt_runtime.h>
#include <nxt_main_process.h>
#include <nxt_conf.h>
#include <nxt_cert.h>


typedef struct {
    nxt_conf_value_t  *root;
    nxt_mp_t          *pool;
} nxt_controller_conf_t;


typedef struct {
    nxt_http_request_parse_t  parser;
    size_t                    length;
    nxt_controller_conf_t     conf;
    nxt_conn_t                *conn;
    nxt_queue_link_t          link;
} nxt_controller_request_t;


typedef struct {
    nxt_uint_t        status;
    nxt_conf_value_t  *conf;

    u_char            *title;
    nxt_str_t         detail;
    ssize_t           offset;
    nxt_uint_t        line;
    nxt_uint_t        column;
} nxt_controller_response_t;


static nxt_int_t nxt_controller_prefork(nxt_task_t *task,
    nxt_process_t *process, nxt_mp_t *mp);
static nxt_int_t nxt_controller_start(nxt_task_t *task,
    nxt_process_data_t *data);
static void nxt_controller_process_new_port_handler(nxt_task_t *task,
    nxt_port_recv_msg_t *msg);
static void nxt_controller_send_current_conf(nxt_task_t *task);
static void nxt_controller_router_ready_handler(nxt_task_t *task,
    nxt_port_recv_msg_t *msg);
static void nxt_controller_remove_pid_handler(nxt_task_t *task,
    nxt_port_recv_msg_t *msg);
static nxt_int_t nxt_controller_conf_default(void);
static void nxt_controller_conf_init_handler(nxt_task_t *task,
    nxt_port_recv_msg_t *msg, void *data);
static void nxt_controller_flush_requests(nxt_task_t *task);
static nxt_int_t nxt_controller_conf_send(nxt_task_t *task, nxt_mp_t *mp,
    nxt_conf_value_t *conf, nxt_port_rpc_handler_t handler, void *data);

static void nxt_controller_conn_init(nxt_task_t *task, void *obj, void *data);
static void nxt_controller_conn_read(nxt_task_t *task, void *obj, void *data);
static nxt_msec_t nxt_controller_conn_timeout_value(nxt_conn_t *c,
    uintptr_t data);
static void nxt_controller_conn_read_error(nxt_task_t *task, void *obj,
    void *data);
static void nxt_controller_conn_read_timeout(nxt_task_t *task, void *obj,
    void *data);
static void nxt_controller_conn_body_read(nxt_task_t *task, void *obj,
    void *data);
static void nxt_controller_conn_write(nxt_task_t *task, void *obj, void *data);
static void nxt_controller_conn_write_error(nxt_task_t *task, void *obj,
    void *data);
static void nxt_controller_conn_write_timeout(nxt_task_t *task, void *obj,
    void *data);
static void nxt_controller_conn_close(nxt_task_t *task, void *obj, void *data);
static void nxt_controller_conn_free(nxt_task_t *task, void *obj, void *data);

static nxt_int_t nxt_controller_request_content_length(void *ctx,
    nxt_http_field_t *field, uintptr_t data);

static void nxt_controller_process_request(nxt_task_t *task,
    nxt_controller_request_t *req);
static void nxt_controller_process_config(nxt_task_t *task,
    nxt_controller_request_t *req, nxt_str_t *path);
static nxt_bool_t nxt_controller_check_postpone_request(nxt_task_t *task);
#if (NXT_TLS)
static void nxt_controller_process_cert(nxt_task_t *task,
    nxt_controller_request_t *req, nxt_str_t *path);
static void nxt_controller_process_cert_save(nxt_task_t *task,
    nxt_port_recv_msg_t *msg, void *data);
static nxt_bool_t nxt_controller_cert_in_use(nxt_str_t *name);
static void nxt_controller_cert_cleanup(nxt_task_t *task, void *obj,
    void *data);
#endif
static void nxt_controller_conf_handler(nxt_task_t *task,
    nxt_port_recv_msg_t *msg, void *data);
static void nxt_controller_conf_store(nxt_task_t *task,
    nxt_conf_value_t *conf);
static void nxt_controller_response(nxt_task_t *task,
    nxt_controller_request_t *req, nxt_controller_response_t *resp);
static u_char *nxt_controller_date(u_char *buf, nxt_realtime_t *now,
    struct tm *tm, size_t size, const char *format);


static nxt_http_field_proc_t  nxt_controller_request_fields[] = {
    { nxt_string("Content-Length"),
      &nxt_controller_request_content_length, 0 },
};

static nxt_lvlhsh_t            nxt_controller_fields_hash;

static nxt_uint_t              nxt_controller_listening;
static nxt_uint_t              nxt_controller_router_ready;
static nxt_controller_conf_t   nxt_controller_conf;
static nxt_queue_t             nxt_controller_waiting_requests;
static nxt_bool_t              nxt_controller_waiting_init_conf;


static const nxt_event_conn_state_t  nxt_controller_conn_read_state;
static const nxt_event_conn_state_t  nxt_controller_conn_body_read_state;
static const nxt_event_conn_state_t  nxt_controller_conn_write_state;
static const nxt_event_conn_state_t  nxt_controller_conn_close_state;


static const nxt_port_handlers_t  nxt_controller_process_port_handlers = {
    .quit           = nxt_signal_quit_handler,
    .new_port       = nxt_controller_process_new_port_handler,
    .change_file    = nxt_port_change_log_file_handler,
    .mmap           = nxt_port_mmap_handler,
    .process_ready  = nxt_controller_router_ready_handler,
    .data           = nxt_port_data_handler,
    .remove_pid     = nxt_controller_remove_pid_handler,
    .rpc_ready      = nxt_port_rpc_handler,
    .rpc_error      = nxt_port_rpc_handler,
};


const nxt_process_init_t  nxt_controller_process = {
    .name           = "controller",
    .type           = NXT_PROCESS_CONTROLLER,
    .prefork        = nxt_controller_prefork,
    .restart        = 1,
    .setup          = nxt_process_core_setup,
    .start          = nxt_controller_start,
    .port_handlers  = &nxt_controller_process_port_handlers,
    .signals        = nxt_process_signals,
};


static nxt_int_t
nxt_controller_prefork(nxt_task_t *task, nxt_process_t *process, nxt_mp_t *mp)
{
    ssize_t                n;
    nxt_int_t              ret;
    nxt_str_t              *conf;
    nxt_file_t             file;
    nxt_runtime_t          *rt;
    nxt_file_info_t        fi;
    nxt_controller_init_t  ctrl_init;

    nxt_log(task, NXT_LOG_INFO, "controller started");

    rt = task->thread->runtime;

    nxt_memzero(&ctrl_init, sizeof(nxt_controller_init_t));

    conf = &ctrl_init.conf;

    nxt_memzero(&file, sizeof(nxt_file_t));

    file.name = (nxt_file_name_t *) rt->conf;

    ret = nxt_file_open(task, &file, NXT_FILE_RDONLY, NXT_FILE_OPEN, 0);

    if (ret == NXT_OK) {
        ret = nxt_file_info(&file, &fi);

        if (nxt_fast_path(ret == NXT_OK && nxt_is_file(&fi))) {
            conf->length = nxt_file_size(&fi);
            conf->start = nxt_mp_alloc(mp, conf->length);
            if (nxt_slow_path(conf->start == NULL)) {
                nxt_file_close(task, &file);
                return NXT_ERROR;
            }

            n = nxt_file_read(&file, conf->start, conf->length, 0);

            if (nxt_slow_path(n != (ssize_t) conf->length)) {
                conf->start = NULL;
                conf->length = 0;

                nxt_alert(task, "failed to restore previous configuration: "
                          "cannot read the file");
            }
        }

        nxt_file_close(task, &file);
    }

#if (NXT_TLS)
    ctrl_init.certs = nxt_cert_store_load(task, mp);

    nxt_mp_cleanup(mp, nxt_controller_cert_cleanup, task, ctrl_init.certs, rt);
#endif

    process->data.controller = ctrl_init;

    return NXT_OK;
}


#if (NXT_TLS)

static void
nxt_controller_cert_cleanup(nxt_task_t *task, void *obj, void *data)
{
    pid_t          main_pid;
    nxt_array_t    *certs;
    nxt_runtime_t  *rt;

    certs = obj;
    rt = data;

    main_pid = rt->port_by_type[NXT_PROCESS_MAIN]->pid;

    if (nxt_pid == main_pid && certs != NULL) {
        nxt_cert_store_release(certs);
    }
}

#endif


static nxt_int_t
nxt_controller_start(nxt_task_t *task, nxt_process_data_t *data)
{
    nxt_mp_t               *mp;
    nxt_int_t              ret;
    nxt_str_t              *json;
    nxt_conf_value_t       *conf;
    nxt_conf_validation_t  vldt;
    nxt_controller_init_t  *init;

    ret = nxt_http_fields_hash(&nxt_controller_fields_hash,
                               nxt_controller_request_fields,
                               nxt_nitems(nxt_controller_request_fields));

    if (nxt_slow_path(ret != NXT_OK)) {
        return NXT_ERROR;
    }

    nxt_queue_init(&nxt_controller_waiting_requests);

    init = &data->controller;

#if (NXT_TLS)
    if (init->certs != NULL) {
        nxt_cert_info_init(task, init->certs);
        nxt_cert_store_release(init->certs);
    }
#endif

    json = &init->conf;

    if (json->start == NULL) {
        return NXT_OK;
    }

    mp = nxt_mp_create(1024, 128, 256, 32);
    if (nxt_slow_path(mp == NULL)) {
        return NXT_ERROR;
    }

    conf = nxt_conf_json_parse_str(mp, json);
    if (nxt_slow_path(conf == NULL)) {
        nxt_alert(task, "failed to restore previous configuration: "
                  "file is corrupted or not enough memory");

        nxt_mp_destroy(mp);
        return NXT_OK;
    }

    nxt_memzero(&vldt, sizeof(nxt_conf_validation_t));

    vldt.pool = nxt_mp_create(1024, 128, 256, 32);
    if (nxt_slow_path(vldt.pool == NULL)) {
        nxt_mp_destroy(mp);
        return NXT_ERROR;
    }

    vldt.conf = conf;

    ret = nxt_conf_validate(&vldt);

    if (nxt_slow_path(ret != NXT_OK)) {

        if (ret == NXT_DECLINED) {
            nxt_alert(task, "the previous configuration is invalid: %V",
                      &vldt.error);

            nxt_mp_destroy(vldt.pool);
            nxt_mp_destroy(mp);

            return NXT_OK;
        }

        /* ret == NXT_ERROR */

        return NXT_ERROR;
    }

    nxt_mp_destroy(vldt.pool);

    nxt_controller_conf.root = conf;
    nxt_controller_conf.pool = mp;

    return NXT_OK;
}


static void
nxt_controller_process_new_port_handler(nxt_task_t *task,
    nxt_port_recv_msg_t *msg)
{
    nxt_port_new_port_handler(task, msg);

    if (msg->u.new_port->type != NXT_PROCESS_ROUTER
        || !nxt_controller_router_ready)
    {
        return;
    }

    nxt_controller_send_current_conf(task);
}


static void
nxt_controller_send_current_conf(nxt_task_t *task)
{
    nxt_int_t         rc;
    nxt_runtime_t     *rt;
    nxt_conf_value_t  *conf;

    conf = nxt_controller_conf.root;

    if (conf != NULL) {
        rc = nxt_controller_conf_send(task, nxt_controller_conf.pool, conf,
                                      nxt_controller_conf_init_handler, NULL);

        if (nxt_fast_path(rc == NXT_OK)) {
            nxt_controller_waiting_init_conf = 1;

            return;
        }

        nxt_mp_destroy(nxt_controller_conf.pool);

        if (nxt_slow_path(nxt_controller_conf_default() != NXT_OK)) {
            nxt_abort();
        }
    }

    if (nxt_slow_path(nxt_controller_conf_default() != NXT_OK)) {
        nxt_abort();
    }

    rt = task->thread->runtime;

    if (nxt_slow_path(nxt_listen_event(task, rt->controller_socket) == NULL)) {
        nxt_abort();
    }

    nxt_controller_listening = 1;

    nxt_controller_flush_requests(task);
}


static void
nxt_controller_router_ready_handler(nxt_task_t *task,
    nxt_port_recv_msg_t *msg)
{
    nxt_port_t     *router_port;
    nxt_runtime_t  *rt;

    rt = task->thread->runtime;

    router_port = rt->port_by_type[NXT_PROCESS_ROUTER];

    nxt_controller_router_ready = 1;

    if (router_port != NULL) {
        nxt_controller_send_current_conf(task);
    }
}


static void
nxt_controller_remove_pid_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg)
{
    nxt_pid_t      pid;
    nxt_process_t  *process;
    nxt_runtime_t  *rt;

    rt = task->thread->runtime;

    nxt_assert(nxt_buf_used_size(msg->buf) == sizeof(pid));

    nxt_memcpy(&pid, msg->buf->mem.pos, sizeof(pid));

    process = nxt_runtime_process_find(rt, pid);
    if (process != NULL && nxt_process_type(process) == NXT_PROCESS_ROUTER) {
        nxt_controller_router_ready = 0;
    }

    nxt_port_remove_pid_handler(task, msg);
}


static nxt_int_t
nxt_controller_conf_default(void)
{
    nxt_mp_t          *mp;
    nxt_conf_value_t  *conf;

    static const nxt_str_t json
        = nxt_string("{ \"listeners\": {}, \"applications\": {} }");

    mp = nxt_mp_create(1024, 128, 256, 32);

    if (nxt_slow_path(mp == NULL)) {
        return NXT_ERROR;
    }

    conf = nxt_conf_json_parse_str(mp, &json);

    if (nxt_slow_path(conf == NULL)) {
        return NXT_ERROR;
    }

    nxt_controller_conf.root = conf;
    nxt_controller_conf.pool = mp;

    return NXT_OK;
}


static void
nxt_controller_conf_init_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
    void *data)
{
    nxt_runtime_t  *rt;

    nxt_controller_waiting_init_conf = 0;

    if (msg->port_msg.type != NXT_PORT_MSG_RPC_READY) {
        nxt_alert(task, "failed to apply previous configuration");

        nxt_mp_destroy(nxt_controller_conf.pool);

        if (nxt_slow_path(nxt_controller_conf_default() != NXT_OK)) {
            nxt_abort();
        }
    }

    if (nxt_controller_listening == 0) {
        rt = task->thread->runtime;

        if (nxt_slow_path(nxt_listen_event(task, rt->controller_socket)
                          == NULL))
        {
            nxt_abort();
        }

        nxt_controller_listening = 1;
    }

    nxt_controller_flush_requests(task);
}


static void
nxt_controller_flush_requests(nxt_task_t *task)
{
    nxt_queue_t               queue;
    nxt_controller_request_t  *req;

    nxt_queue_init(&queue);
    nxt_queue_add(&queue, &nxt_controller_waiting_requests);

    nxt_queue_init(&nxt_controller_waiting_requests);

    nxt_queue_each(req, &queue, nxt_controller_request_t, link) {
        nxt_controller_process_request(task, req);
    } nxt_queue_loop;
}


static nxt_int_t
nxt_controller_conf_send(nxt_task_t *task, nxt_mp_t *mp, nxt_conf_value_t *conf,
    nxt_port_rpc_handler_t handler, void *data)
{
    void           *mem;
    u_char         *end;
    size_t         size;
    uint32_t       stream;
    nxt_fd_t       fd;
    nxt_int_t      rc;
    nxt_buf_t      *b;
    nxt_port_t     *router_port, *controller_port;
    nxt_runtime_t  *rt;

    rt = task->thread->runtime;

    router_port = rt->port_by_type[NXT_PROCESS_ROUTER];

    nxt_assert(router_port != NULL);
    nxt_assert(nxt_controller_router_ready);

    controller_port = rt->port_by_type[NXT_PROCESS_CONTROLLER];

    size = nxt_conf_json_length(conf, NULL);

    b = nxt_buf_mem_alloc(mp, sizeof(size_t), 0);
    if (nxt_slow_path(b == NULL)) {
        return NXT_ERROR;
    }

    fd = nxt_shm_open(task, size);
    if (nxt_slow_path(fd == -1)) {
        return NXT_ERROR;
    }

    mem = nxt_mem_mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
    if (nxt_slow_path(mem == MAP_FAILED)) {
        goto fail;
    }

    end = nxt_conf_json_print(mem, conf, NULL);

    nxt_mem_munmap(mem, size);

    size = end - (u_char *) mem;

    b->mem.free = nxt_cpymem(b->mem.pos, &size, sizeof(size_t));

    stream = nxt_port_rpc_register_handler(task, controller_port,
                                           handler, handler,
                                           router_port->pid, data);
    if (nxt_slow_path(stream == 0)) {
        goto fail;
    }

    rc = nxt_port_socket_write(task, router_port,
                               NXT_PORT_MSG_DATA_LAST | NXT_PORT_MSG_CLOSE_FD,
                               fd, stream, controller_port->id, b);

    if (nxt_slow_path(rc != NXT_OK)) {
        nxt_port_rpc_cancel(task, controller_port, stream);

        goto fail;
    }

    return NXT_OK;

fail:

    nxt_fd_close(fd);

    return NXT_ERROR;
}


nxt_int_t
nxt_runtime_controller_socket(nxt_task_t *task, nxt_runtime_t *rt)
{
    nxt_listen_socket_t  *ls;

    ls = nxt_mp_alloc(rt->mem_pool, sizeof(nxt_listen_socket_t));
    if (ls == NULL) {
        return NXT_ERROR;
    }

    ls->sockaddr = rt->controller_listen;

    nxt_listen_socket_remote_size(ls);

    ls->socket = -1;
    ls->backlog = NXT_LISTEN_BACKLOG;
    ls->read_after_accept = 1;
    ls->flags = NXT_NONBLOCK;

#if 0
    /* STUB */
    wq = nxt_mp_zget(cf->mem_pool, sizeof(nxt_work_queue_t));
    if (wq == NULL) {
        return NXT_ERROR;
    }
    nxt_work_queue_name(wq, "listen");
    /**/

    ls->work_queue = wq;
#endif
    ls->handler = nxt_controller_conn_init;

    if (nxt_listen_socket_create(task, rt->mem_pool, ls) != NXT_OK) {
        return NXT_ERROR;
    }

    rt->controller_socket = ls;

    return NXT_OK;
}


static void
nxt_controller_conn_init(nxt_task_t *task, void *obj, void *data)
{
    nxt_buf_t                 *b;
    nxt_conn_t                *c;
    nxt_event_engine_t        *engine;
    nxt_controller_request_t  *r;

    c = obj;

    nxt_debug(task, "controller conn init fd:%d", c->socket.fd);

    r = nxt_mp_zget(c->mem_pool, sizeof(nxt_controller_request_t));
    if (nxt_slow_path(r == NULL)) {
        nxt_controller_conn_free(task, c, NULL);
        return;
    }

    r->conn = c;

    if (nxt_slow_path(nxt_http_parse_request_init(&r->parser, c->mem_pool)
                      != NXT_OK))
    {
        nxt_controller_conn_free(task, c, NULL);
        return;
    }

    r->parser.encoded_slashes = 1;

    b = nxt_buf_mem_alloc(c->mem_pool, 1024, 0);
    if (nxt_slow_path(b == NULL)) {
        nxt_controller_conn_free(task, c, NULL);
        return;
    }

    c->read = b;
    c->socket.data = r;
    c->socket.read_ready = 1;
    c->read_state = &nxt_controller_conn_read_state;

    engine = task->thread->engine;
    c->read_work_queue = &engine->read_work_queue;
    c->write_work_queue = &engine->write_work_queue;

    nxt_conn_read(engine, c);
}


static const nxt_event_conn_state_t  nxt_controller_conn_read_state
    nxt_aligned(64) =
{
    .ready_handler = nxt_controller_conn_read,
    .close_handler = nxt_controller_conn_close,
    .error_handler = nxt_controller_conn_read_error,

    .timer_handler = nxt_controller_conn_read_timeout,
    .timer_value = nxt_controller_conn_timeout_value,
    .timer_data = 60 * 1000,
};


static void
nxt_controller_conn_read(nxt_task_t *task, void *obj, void *data)
{
    size_t                    preread;
    nxt_buf_t                 *b;
    nxt_int_t                 rc;
    nxt_conn_t                *c;
    nxt_controller_request_t  *r;

    c = obj;
    r = data;

    nxt_debug(task, "controller conn read");

    nxt_queue_remove(&c->link);
    nxt_queue_self(&c->link);

    b = c->read;

    rc = nxt_http_parse_request(&r->parser, &b->mem);