path: root/wsrep-lib/src/reporter.cpp

/*
 * Copyright (C) 2021 Codership Oy <info@codership.com>
 *
 * This file is part of wsrep-lib.
 *
 * Wsrep-lib is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 2 of the License, or
 * (at your option) any later version.
 *
 * Wsrep-lib is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with wsrep-lib.  If not, see <https://www.gnu.org/licenses/>.
 */

#include "wsrep/reporter.hpp"
#include "wsrep/logger.hpp"

#include <sstream>
#include <iomanip>

#include <cassert>
#include <cstring>  // strerror()
#include <cstdlib>  // mkstemp()
#include <cerrno>   // errno
#include <unistd.h> // write()
#include <cstdio>   // rename(), snprintf()
#include <ctime>    // clock_gettime()
#include <cmath>    // floor()

static std::string const TEMP_EXTENSION(".XXXXXX");

static std::string make_progress_string(int const from, int const to,
                                        int const total,int const done,
                                        int const indefinite)
{
    std::ostringstream os;

    os << "{ \"from\": "     << from       << ", "
       << "\"to\": "         << to         << ", "
       << "\"total\": "      << total      << ", "
       << "\"done\": "       << done       << ", "
       << "\"indefinite\": " << indefinite << " }";

    return os.str();
}

static std::string const indefinite_progress
    (make_progress_string(-1, -1, -1, -1, -1));
static std::string const steady_state
    (make_progress_string(-1, -1, 0, 0, -1));

static inline double
timestamp()
{
    struct timespec time;
    clock_gettime(CLOCK_REALTIME, &time);
    return (double(time.tv_sec) + double(time.tv_nsec)*1.0e-9);
}

wsrep::reporter::reporter(wsrep::mutex&      mutex,
                          const std::string& file_name,
                          size_t const       max_msg)
    : mutex_(mutex)
    , file_name_(file_name)
    , progress_(indefinite_progress)
    , template_(new char [file_name_.length() + TEMP_EXTENSION.length() + 1])
    , state_(wsrep::reporter::s_disconnected_disconnected)
    , initialized_(false)
    , err_msg_()
    , warn_msg_()
    , events_()
    , max_msg_(max_msg)
{
    template_[file_name_.length() + TEMP_EXTENSION.length()] = '\0';
    write_file(timestamp());
}

wsrep::reporter::~reporter()
{
    delete [] template_;
}

wsrep::reporter::substates
wsrep::reporter::substate_map(enum wsrep::server_state::state const state)
{
    switch (state)
    {
    case wsrep::server_state::s_disconnected:
        initialized_ = false;
        return s_disconnected_disconnected;
    case wsrep::server_state::s_initializing:
        if (s_disconnected_disconnected == state_)
            return s_disconnected_initializing;
        else if (s_joining_sst == state_)
            return s_joining_initializing;
        else if (s_joining_initializing == state_)
            return s_joining_initializing; // continuation
        else
        {
            assert(0);
            return state_;
        }
    case wsrep::server_state::s_initialized:
        initialized_ = true;
        if (s_disconnected_initializing >= state_)
            return s_disconnected_initialized;
        else if (s_joining_initializing == state_)
            return s_joining_ist;
        else if (s_joining_ist == state_)
            return s_joining_ist; // continuation
        else
        {
            assert(0);
            return state_;
        }
    case wsrep::server_state::s_connected:
        return s_connected_waiting;
    case wsrep::server_state::s_joiner:
        if (initialized_)
            return s_joining_initialized;
        else
            return s_joining_sst;
    case wsrep::server_state::s_joined:
        return s_joined_syncing;
    case wsrep::server_state::s_donor:
        return s_donor_sending;
    case wsrep::server_state::s_synced:
        return s_synced_running;
    case wsrep::server_state::s_disconnecting:
        return s_disconnecting_disconnecting;
    default:
        assert(0);
        return state_;
    }
}

// See https://www.ietf.org/rfc/rfc4627.txt
static std::string escape_json(const std::string& str)
{
    std::ostringstream os;
    for (auto c = str.cbegin(); c != str.cend(); ++c)
    {
        switch (*c)
        {
        case '"': os << "\\\""; break;
        case '\\': os << "\\\\"; break;
        case '/': os << "\\/"; break;
        case '\b': os << "\\b"; break;
        case '\f': os << "\\f"; break;
        case '\n': os << "\\n"; break;
        case '\r': os << "\\r"; break;
        case '\t': os << "\\t"; break;
        default:
            // std::iscntrl() returns non-zero for [0x00, 0x1f] and
            // backspace character. Argument type is int so cast to
            // unsigned char is needed to make it safe, see
            // https://en.cppreference.com/w/cpp/string/byte/iscntrl
            if (std::iscntrl(static_cast<unsigned char>(*c)))
            {
                os << "\\u" << std::hex << std::setw(4) <<
                    std::setfill('0') << static_cast<int>(*c);
            }
            else
            {
                os << *c;
            }
        }
    }
    return os.str();
}

void
wsrep::reporter::write_log_msg(std::ostream&  os,
                               const log_msg& msg)
{
    os << "\t\t{\n";
    os << "\t\t\t\"timestamp\": " << std::showpoint << std::setprecision(18)
       << msg.tstamp << ",\n";
    os << "\t\t\t\"msg\": \"" << msg.msg << "\"\n";
    os << "\t\t}";
}

void
wsrep::reporter::write_event(std::ostream&  os,
                             const log_msg& msg)
{
    os << "\t\t{\n";
    os << "\t\t\t\"timestamp\": " << std::showpoint << std::setprecision(18)
       << msg.tstamp << ",\n";
    os << "\t\t\t\"event\": " << msg.msg << "\n";
    os << "\t\t}";
}

void
wsrep::reporter::write_array(std::ostream&              os,
                             const std::string&         label,
                             const std::deque<log_msg>& msgs,
                             void (*element_writer)(std::ostream& os,
                                                    const log_msg& msg))
{
    os << "\t\"" << label << "\": [\n";
    for (size_t i(0); i < msgs.size(); ++i)
    {
        element_writer(os, msgs[i]);
        os << (i+1 < msgs.size() ? ",\n" : "\n");
    }
    os << "\t],\n";
}

// write data to temporary file and then rename it to target file for atomicity
void
wsrep::reporter::write_file(double const tstamp)
{
    enum progress_type {
        t_indefinite = -1, // indefinite wait
        t_progressive,     // measurable progress
        t_final            // final state
    };

    struct strings {
        const char*   state;
        const char*   comment;
        progress_type type;
    };

    static const struct strings strings[substates_max] =
        {
            { "DISCONNECTED",  "Disconnected",    t_indefinite  },
            { "DISCONNECTED",  "Initializing",    t_indefinite  },
            { "DISCONNECTED",  "Connecting",      t_indefinite  },
            { "CONNECTED",     "Waiting",         t_indefinite  },
            { "JOINING",       "Receiving state", t_progressive },
            { "JOINING",       "Receiving SST",   t_progressive },
            { "JOINING",       "Initializing",    t_progressive },
            { "JOINING",       "Receiving IST",   t_progressive },
            { "JOINED",        "Syncing",         t_progressive },
            { "SYNCED",        "Operational",     t_final       },
            { "DONOR",         "Donating SST",    t_progressive },
            { "DISCONNECTING", "Disconnecting",   t_indefinite  }
        };

    double const seconds(floor(tstamp));
    time_t const tt = time_t(seconds);
    struct tm    date;
    localtime_r(&tt, &date);

    char date_str[85] = { '\0', };
    snprintf(date_str, sizeof(date_str) - 1,
             "%04d-%02d-%02d %02d:%02d:%02d.%03d",
             date.tm_year + 1900, date.tm_mon + 1, date.tm_mday,
             date.tm_hour, date.tm_min, date.tm_sec,
             (int)((tstamp-seconds)*1000));

    std::ostringstream os;
    os << "{\n";
    os << "\t\"date\": \"" << date_str << "\",\n";
    os << "\t\"timestamp\": " << std::showpoint << std::setprecision(18)
       << tstamp << ",\n";
    write_array(os, "errors",   err_msg_, write_log_msg);
    write_array(os, "warnings", warn_msg_, write_log_msg);
    write_array(os, "events", events_, write_event);
    os << "\t\"status\": {\n";
    os << "\t\t\"state\": \"" << strings[state_].state << "\",\n";
    os << "\t\t\"comment\": \"" << strings[state_].comment << "\",\n";
    os << "\t\t\"progress\": " << progress_ << "\n";
    os << "\t}\n";
    os << "}\n";

    std::string const str(os.str());

    // prepare template for mkstemp()
    file_name_.copy(template_, file_name_.length());
    TEMP_EXTENSION.copy(template_ +file_name_.length(),TEMP_EXTENSION.length());

    int const fd(mkstemp(template_));
    if (fd < 0)
    {
        std::cerr << "Reporter could not open temporary file `" << template_
                  << "': " << strerror(errno) << " (" << errno << ")\n";
        return;
    }
    ssize_t err(write(fd, str.c_str(), str.length()));
    close(fd);
    if (err < 0)
    {
        std::cerr << "Could not write " << str.length()
                  << " bytes to temporary file '"
                  << template_ << "': " << strerror(errno)
                  << " (" << errno << ")\n";
        return;
    }

    rename(template_, file_name_.c_str());
}

void
wsrep::reporter::report_state(enum server_state::state const s)
{
    wsrep::unique_lock<wsrep::mutex> lock(mutex_);

    substates const state(substate_map(s));

    if (state != state_)
    {
        state_ = state;

        if (state_ == s_synced_running)
            progress_ = steady_state;
        else
            progress_ = indefinite_progress;

        write_file(timestamp());
    }
}

void
wsrep::reporter::report_progress(const std::string& json)
{
    wsrep::unique_lock<wsrep::mutex> lock(mutex_);

    if (json != progress_)
    {
        if (state_ != s_synced_running)
        {
            // ignore any progress in SYNCED state
            progress_ = json;
            write_file(timestamp());
        }
    }
}

void
wsrep::reporter::report_event(const std::string& json)
{
    wsrep::unique_lock<wsrep::mutex> lock(mutex_);
    if (events_.size() == max_msg_)
    {
        events_.pop_front();
    }
    events_.push_back({timestamp(), json});
    write_file(timestamp());
}

void
wsrep::reporter::report_log_msg(log_level const    lvl,
                                const std::string& msg,
                                double             tstamp)
{
    std::deque<log_msg>& deque(lvl == error ? err_msg_ : warn_msg_);

    wsrep::unique_lock<wsrep::mutex> lock(mutex_);

    if (deque.empty() || msg != deque.back().msg)
    {
        if (deque.size() == max_msg_) deque.pop_front();

        if (tstamp <= undefined) tstamp = timestamp();

        /* Log messages are not expected to be json formatted, so we escape
           the message strings here to keep the report file well formatted. */
        log_msg entry({tstamp, escape_json(msg)});
        deque.push_back(entry);
        write_file(tstamp);
    }
}