AdminDBClient.cpp

Go to the documentation of this file.

#include <cassert>
#include <chrono>
#include <ctime>
#include <log/logger.h>
#include <memory>
#include <sstream>
#include <thread>
 
#include <pqxx/pqxx>
#include <utility>
 
#include <log/macros.h>
 
#include "AdminDBClient.hpp"
#include "common_vars.hpp"
#include "log_admindb_host_reservation_importer.h"
#include "util/pqxx_compat.hpp"
 
namespace
{
STMT_NAME_T STMT_FETCH_HOST_UPDATES{"kch-ahri-fetch-host-updates"};
STMT_NAME_T STMT_FETCH_FULL_DHCP_TABLE{"kch-ahri-fetch-full-dhcp-table"};
STMT_NAME_T STMT_FETCH_MAX_UPDATE_ID{"kch-ahri-fetch-max-update-id"};
STMT_NAME_T STMT_PRUNE_OLD_HOST_UPDATES{"kch-ahri-prune-old-host-updates"};
} // namespace
 
namespace ahri
{
class AdminDBClientNotificationHandler : public pqxx::notification_receiver
{
public:
    AdminDBClientNotificationHandler(pqxx::connection& conn,
                                     AdminDBClient::dhcp_update_callback_t cb)
        : pqxx::notification_receiver(conn, AdminDBClient::NOTIFICATION_CHANNEL_NAME)
        , callback(std::move(cb))
    {}
 
    void operator()([[maybe_unused]] const std::string& payload,
                    [[maybe_unused]] int backend_pid) final
    {
        callback();
    }
 
private:
    AdminDBClient::dhcp_update_callback_t callback;
 
    friend struct AdminDBClientPrivate;
};
 
struct AdminDBClientPrivate
{
    AdminDBClientPrivate(std::string psql_connstring, size_t initial_last_processed_update_id)
        : psql_connstring(std::move(psql_connstring))
        , last_returned_update_id(initial_last_processed_update_id)
    {
        reconnect();
    }
 
    AdminDBClientPrivate(const AdminDBClientPrivate&) = delete;
    AdminDBClientPrivate& operator=(const AdminDBClientPrivate&) = delete;
    AdminDBClientPrivate(AdminDBClientPrivate&&) = delete;
    AdminDBClientPrivate& operator=(AdminDBClientPrivate&&) = delete;
 
    ~AdminDBClientPrivate()
    {
        run_notification_receiver = false;
        std::unique_lock lck{psql_connection_mtx};
        // Better keep this order, it seems to prevent the notification receiver thread not joining
        // properly.
#if KCH_PQXX_MAJOR_VERSION < 7
        psql_connection->disconnect();
#else
        psql_connection->close();
#endif
        active_notification_handler.reset(nullptr);
        if (notification_receiver_thread && notification_receiver_thread->joinable()) {
            notification_receiver_thread->join();
        }
 
        lck.unlock();
    }
 
    void reconnect()
    {
        std::lock_guard lck{psql_connection_mtx};
        psql_connection = std::make_unique<pqxx::connection>(psql_connstring);
 
        // Inner join might cause the result to become unordered, so we need to order explicitly
        psql_connection->prepare(
          STMT_FETCH_HOST_UPDATES,
          "SELECT u.*, r.t_id * 10000 + r.t_vlan_id AS t_kea_subnet_id "
          "FROM admindb_dhcp_updates_get() u "
          "INNER JOIN admindb_get_all_vlan_iprange() r ON u.t_subnet_id = r.t_id "
          "WHERE u.t_id > $1 "
          "ORDER BY u.t_id ASC;");
        psql_connection->prepare(
          STMT_FETCH_FULL_DHCP_TABLE,
          "SELECT v.t_ip, v.t_mac, v.t_subnet_id * 10000 + v.t_vlan_id as t_subnet_id "
          "FROM admindb_view_kea_dhcp() v;");
        psql_connection->prepare(STMT_FETCH_MAX_UPDATE_ID,
                                 "SELECT MAX(t_id) FROM admindb_dhcp_updates_get();");
        psql_connection->prepare(
          STMT_PRUNE_OLD_HOST_UPDATES,
          "SELECT admindb_dhcp_updates_cleanup_before(($1 :: TIMESTAMPTZ) :: TIMESTAMP)");
 
        if (active_notification_handler) {
            handle_notifications_int(active_notification_handler->callback);
        }
    }
 
    void handle_notifications_int(const AdminDBClient::dhcp_update_callback_t& cb)
    {
        active_notification_handler =
          std::make_unique<AdminDBClientNotificationHandler>(*psql_connection, cb);
        start_notifcation_receiver_thread();
    }
 
    void attempt_reconnect()
    {
        while (true) {
            try {
                // Account for failover durations in case one of the AdminDB masters has
                // problems
                std::this_thread::sleep_for(std::chrono::minutes(1));
                reconnect();
                LOG_INFO(logger, AHRI_CONNECTION_RESTORED).arg("AdminDB");
                break;
            }
            catch (pqxx::broken_connection& e) {
                LOG_WARN(logger, AHRI_CONNECTION_FAILURE).arg("AdminDB").arg(e.what());
            }
            catch (pqxx::failure& e) {
                // Before we know how to handle every failure properly, resort to
                // halting operation, keeping the DHCP server operational without
                // causing further issues
                LOG_FATAL(logger, AHRI_PQXX_FAILURE_HALT).arg("AdminDB reconnect").arg(e.what());
                run_notification_receiver = false;
                break;
            }
        }
    }
 
    void start_notifcation_receiver_thread()
    {
        if (notification_receiver_thread && notification_receiver_thread->joinable()) {
            return;
        }
 
        notification_receiver_thread =
          std::make_unique<std::thread>([this]() { notification_receiver_worker(); });
    }
 
    void notification_receiver_worker()
    {
        LOG_DEBUG(logger, 10, AHRI_DEBUG_NOTIFICATION_RECEIVER_STARTED);
        while (active_notification_handler && run_notification_receiver) {
            try {
                std::unique_lock lck{psql_connection_mtx};
                psql_connection->await_notification(
                  std::chrono::duration_cast<std::chrono::seconds>(
                    AdminDBClient::NOTIFICATION_RECEIVER_TIMEOUT)
                    .count(),
                  0);
                lck.unlock();
                std::this_thread::sleep_for(AdminDBClient::NOTIFICATION_RECEIVER_SLEEP);
            }
            catch (pqxx::broken_connection& e) {
                LOG_WARN(logger, AHRI_BROKEN_CONNECTION).arg("AdminDB");
                if (run_notification_receiver) {
                    attempt_reconnect();
                }
            }
            catch (pqxx::failure& e) {
                // Before we know how to handle every failure properly, resort to halting operation,
                // keeping the DHCP server operational without causing further issues
                LOG_FATAL(logger, AHRI_PQXX_FAILURE_HALT)
                  .arg("AdminDB notification receiver")
                  .arg(e.what());
                break;
            }
        }
    }
 
    const std::string psql_connstring;
    std::unique_ptr<pqxx::connection> psql_connection{nullptr};
    std::recursive_mutex psql_connection_mtx;
 
    std::unique_ptr<AdminDBClientNotificationHandler> active_notification_handler;
    bool run_notification_receiver{true};
    std::unique_ptr<std::thread> notification_receiver_thread;
 
    size_t last_returned_update_id{0};
 
    isc::log::Logger logger{"ahri"};
};
 
AdminDBClient::AdminDBClient(std::string_view host, std::string_view port, std::string_view user,
                             std::string_view password, std::string_view database,
                             size_t last_processed_host_update_id)
{
    std::ostringstream connstr;
    connstr << "postgresql://" << user << ':' << password << '@' << host << ':' << port << '/'
            << database;
    p_impl = std::make_unique<AdminDBClientPrivate>(connstr.str(), last_processed_host_update_id);
}
 
AdminDBClient::~AdminDBClient()
{
    p_impl.reset(nullptr);
}
 
void AdminDBClient::handle_notifications(const dhcp_update_callback_t& cb)
{
    std::unique_lock lck{p_impl->psql_connection_mtx};
    p_impl->handle_notifications_int(cb);
}
 
void AdminDBClient::prune_old_host_updates(std::chrono::days prune_before)
{
    const auto prune_date = std::chrono::system_clock::now() - prune_before;
    const std::time_t prune_date_time_t = std::chrono::system_clock::to_time_t(prune_date);
    struct tm t = {};
    gmtime_r(&prune_date_time_t, &t);
    char prune_date_str[] = "yyyy-mm-dd 88:88:88";
    const std::size_t written = strftime(prune_date_str, sizeof(prune_date_str), "%F %T", &t);
    assert(written + 1 == sizeof(prune_date_str));
 
    std::lock_guard lck{p_impl->psql_connection_mtx};
    pqxx::work transaction{*p_impl->psql_connection};
    // Add +00:00 to the timestamp to indicate we are using UTC
    transaction.exec_prepared1(STMT_PRUNE_OLD_HOST_UPDATES,
                               prune_date_str + std::string{" +00:00"});
    transaction.commit();
}
 
std::vector<AdminDBClient::HostUpdate> AdminDBClient::fetch_all_host_reservations()
{
    std::unique_lock lck{p_impl->psql_connection_mtx};
    LOG_DEBUG(p_impl->logger, 50, AHRI_FULL_SYNC_START_TRANSACTION);
    pqxx::work transaction{*p_impl->psql_connection};
 
    LOG_DEBUG(p_impl->logger, 50, AHRI_FULL_SYNC_START_QUERY);
    auto fetch_res = transaction.exec_prepared(STMT_FETCH_FULL_DHCP_TABLE);
    std::vector<HostUpdate> table;
    auto max_incremental_update_id = transaction.exec_prepared1(STMT_FETCH_MAX_UPDATE_ID)
                                       .at(0)
                                       .as<size_t>(BELOW_LOWEST_INCREMENTAL_UPDATE_ID);
 
    LOG_DEBUG(p_impl->logger, 50, AHRI_FULL_SYNC_BUILD_RESERVATIONS);
    for (const auto& row : fetch_res) {
        HostUpdate::Type update_type = HostUpdate::Type::ADD;
        auto ip = isc::asiolink::IOAddress(row.at("t_ip").as<std::string>());
        auto mac = isc::dhcp::HWAddr::fromText(row.at("t_mac").as<std::string>());
        auto subnet_id = row.at("t_subnet_id").as<size_t>(UNDEFINED_SUBNET_ID);
 
        if (ip.isV6()) {
            LOG_DEBUG(p_impl->logger, 70, AHRI_DEBUG_IGNORE_HOST_RESERVATION)
              .arg(mac.toText(false))
              .arg(ip.toText())
              .arg("IPv6 address");
            continue;
        }
        assert(ip.isV4());
 
        table.push_back(
          {update_type, std::move(ip), std::move(mac), subnet_id, max_incremental_update_id});
    }
 
    p_impl->last_returned_update_id = max_incremental_update_id;
 
    LOG_DEBUG(p_impl->logger, 50, AHRI_FULL_SYNC_COMMIT_TRANSACTION);
    transaction.commit();
    return table;
}
 
std::vector<AdminDBClient::HostUpdate> AdminDBClient::fetch_pending_host_updates()
{
    std::unique_lock lck{p_impl->psql_connection_mtx};
    pqxx::work transaction{*p_impl->psql_connection};
 
    auto fetch_res =
      transaction.exec_prepared(STMT_FETCH_HOST_UPDATES, p_impl->last_returned_update_id);
    std::vector<HostUpdate> updates;
    size_t last_update_id = p_impl->last_returned_update_id;
 
    for (const auto& row : fetch_res) {
        auto update_id = row.at("t_id").as<size_t>();
        HostUpdate::Type update_type =
          host_update_type_from_string(row.at("t_action").as<std::string>());
        auto ip = isc::asiolink::IOAddress(row.at("t_ip").as<std::string>());
        auto mac = isc::dhcp::HWAddr::fromText(row.at("t_mac").as<std::string>());
        auto subnet_id = row.at("t_kea_subnet_id").as<size_t>();
 
        if (ip.isV6()) {
            LOG_DEBUG(p_impl->logger, 70, AHRI_DEBUG_IGNORE_HOST_RESERVATION)
              .arg(mac.toText(false))
              .arg(ip.toText())
              .arg("IPv6 address");
            continue;
        }
        assert(ip.isV4());
 
        assert(update_id > last_update_id);
        updates.push_back({update_type, std::move(ip), std::move(mac), subnet_id, update_id});
        last_update_id = update_id;
    }
 
    p_impl->last_returned_update_id = last_update_id;
 
    transaction.commit();
    return updates;
}
 
size_t AdminDBClient::fetch_latest_available_update_id()
{
    std::unique_lock lck{p_impl->psql_connection_mtx};
    pqxx::work transaction{*p_impl->psql_connection};
    auto res = transaction.exec_prepared1(STMT_FETCH_MAX_UPDATE_ID);
 
    transaction.commit();
    return res.at(0).as<size_t>();
}
} // namespace ahri