LocalDBClient.cpp

Go to the documentation of this file.

#include <sstream>
 
#include <pqxx/pqxx>
 
#include <asiolink/io_address.h>
#include <dhcp/hwaddr.h>
 
#include <log/logger.h>
#include <log/macros.h>
 
#include "AdminDBClient.hpp"
#include "LocalDBClient.hpp"
#include "common_vars.hpp"
#include "util/bytea_utils.hpp"
 
#include "log_admindb_host_reservation_importer.h"
#include "util/pqxx_compat.hpp"
 
namespace
{
STMT_NAME_T STMT_INSERT_HOST_RESERVATION{"insert_host_reservation"};
STMT_NAME_T STMT_DELETE_HOST_RESERVATION{"delete_host_reservation"};
STMT_NAME_T STMT_QUERY_HOST_RESERVATION_BY_MAC{"query_host_reservation_by_mac"};
STMT_NAME_T STMT_QUERY_HOST_RESERVATION_BY_IP{"query_host_reservation_by_ip"};
STMT_NAME_T STMT_INSERT_TMP_HOST_RESERVATION{"insert_tmp_host_reservation"};
STMT_NAME_T STMT_CREATE_TMP_HOST_TABLE{"create_tmp_host_table"};
STMT_NAME_T STMT_FLUSH_TMP_HOST_TABLE{"flush_tmp_host_table"};
} // namespace
 
namespace ahri
{
struct LocalDBClientPrivate
{
    LocalDBClientPrivate(const std::string& connstr) : psql(connstr)
    {
        std::ostringstream stmt_insert_host_reservation;
        stmt_insert_host_reservation << "INSERT INTO hosts (dhcp_identifier, dhcp_identifier_type, "
                                        "ipv4_address, dhcp4_subnet_id) "
                                     << "VALUES ($1, " << LocalDBClient::DHCP_IDENTIFIER_TYPE_MAC
                                     << ", $2, $3);";
        std::string stmt_insert_host_reservation_str = stmt_insert_host_reservation.str();
        psql.prepare(STMT_INSERT_HOST_RESERVATION, stmt_insert_host_reservation_str);
 
        std::string stmt_delete_host_reservation{
          "DELETE FROM hosts WHERE dhcp_identifier = $1 AND ipv4_address = $2 AND dhcp4_subnet_id "
          "= $3;"};
        psql.prepare(STMT_DELETE_HOST_RESERVATION, stmt_delete_host_reservation);
 
        const std::string stmt_query_host_reservation_by_ip{
          "SELECT dhcp_identifier, dhcp_identifier_type, ipv4_address, dhcp4_subnet_id FROM hosts "
          "WHERE ipv4_address = $1;"};
        psql.prepare(STMT_QUERY_HOST_RESERVATION_BY_IP, stmt_query_host_reservation_by_ip);
        const std::string stmt_query_host_reservation_by_mac{
          "SELECT dhcp_identifier, dhcp_identifier_type, ipv4_address, dhcp4_subnet_id FROM hosts "
          "WHERE dhcp_identifier = $1;"};
        psql.prepare(STMT_QUERY_HOST_RESERVATION_BY_MAC, stmt_query_host_reservation_by_mac);
 
        std::string stmt_create_tmp_host_table{
          "CREATE TEMPORARY TABLE IF NOT EXISTS tmp_full_host_reservations("
          "id SERIAL PRIMARY KEY NOT NULL,"
          "mac_address BYTEA NOT NULL,"
          "ipv4_address BIGINT NOT NULL,"
          "subnet_id INT NOT NULL);"};
        psql.prepare(STMT_CREATE_TMP_HOST_TABLE, stmt_create_tmp_host_table);
 
        std::string stmt_flush_tmp_host_table{"TRUNCATE tmp_full_host_reservations;"};
        psql.prepare(STMT_FLUSH_TMP_HOST_TABLE, stmt_flush_tmp_host_table);
    }
    LocalDBClientPrivate(const LocalDBClientPrivate&) = delete;
    ~LocalDBClientPrivate() = default;
 
    void apply_host_update(pqxx::work& transaction, const AdminDBClient::HostUpdate& host_update)
    {
#if KCH_PQXX_MAJOR_VERSION < 7
        const pqxx::binarystring dhcp_identifier{util::to_binstr(host_update.mac.hwaddr_)};
#else
        const util::byte_string dhcp_identifier{util::to_bytestr(host_update.mac.hwaddr_)};
#endif
        assert(host_update.ipv4.isV4());
        const uint32_t ipv4_address{host_update.ipv4.toUint32()};
 
        if (host_update.type == AdminDBClient::HostUpdate::Type::ADD)
            transaction.exec_prepared(STMT_INSERT_HOST_RESERVATION, dhcp_identifier, ipv4_address,
                                      host_update.subnet_id);
        else
            transaction.exec_prepared(STMT_DELETE_HOST_RESERVATION, dhcp_identifier, ipv4_address,
                                      host_update.subnet_id);
    }
 
    void setup_tmp_host_table()
    {
        pqxx::work table_transaction{psql};
        table_transaction.exec_prepared0(STMT_CREATE_TMP_HOST_TABLE);
        table_transaction.exec_prepared0(STMT_FLUSH_TMP_HOST_TABLE);
        table_transaction.commit();
    }
 
    void prepare_insert_tmp_host_reservations()
    {
        // The statement must be prepared before the transaction for inserting host reservations is
        // created
        std::string stmt_insert_tmp_host_reservation{
          "INSERT INTO tmp_full_host_reservations "
          "  (mac_address, ipv4_address, subnet_id) "
          "VALUES "
          "  ($1, $2, $3);"};
        psql.prepare(STMT_INSERT_TMP_HOST_RESERVATION, stmt_insert_tmp_host_reservation);
    }
 
    void insert_tmp_host_reservations(pqxx::work& transaction,
                                      const std::vector<AdminDBClient::HostUpdate>& full_sync)
    {
        for (const auto& sync_item : full_sync) {
#if KCH_PQXX_MAJOR_VERSION < 7
            const bytestr_t mac_bin{util::to_binstr(sync_item.mac.hwaddr_)};
#else
            const bytestr_t mac_bin{util::to_bytestr(sync_item.mac.hwaddr_)};
#endif
            transaction.exec_prepared(STMT_INSERT_TMP_HOST_RESERVATION, mac_bin,
                                      sync_item.ipv4.toUint32(), sync_item.subnet_id);
        }
 
        psql.unprepare(STMT_INSERT_TMP_HOST_RESERVATION);
    }
 
    auto compare_current_and_tmp_host_reservations(pqxx::work& transaction)
    {
        // ORDER BY is important to ensure that removals come first in the diff, thus avoiding
        // UNIQUE constraint violations when applying following ADD instructions
        std::string query_compare_host_tables{
          "SELECT"
          "  hosts.host_id AS l_id,"
          "  hosts.dhcp_identifier AS l_mac,"
          "  hosts.ipv4_address AS l_ip,"
          "  hosts.dhcp4_subnet_id AS l_subnet,"
          "  tmp_full_host_reservations.id AS r_id,"
          "  tmp_full_host_reservations.mac_address AS r_mac,"
          "  tmp_full_host_reservations.ipv4_address AS r_ip, "
          "  tmp_full_host_reservations.subnet_id AS r_subnet "
          "FROM hosts "
          "FULL OUTER JOIN tmp_full_host_reservations "
          "ON hosts.dhcp_identifier = tmp_full_host_reservations.mac_address "
          "AND hosts.ipv4_address = tmp_full_host_reservations.ipv4_address "
          "AND hosts.dhcp4_subnet_id = tmp_full_host_reservations.subnet_id "
          "WHERE hosts.host_id IS NULL "
          "OR tmp_full_host_reservations.id IS NULL "
          "ORDER BY l_id ASC "
          ";"};
        return transaction.exec(query_compare_host_tables);
    }
 
    static AdminDBClient::HostUpdate make_host_update_from_diff_row(const pqxx::row& row)
    {
        if (row.at("r_id").is_null()) {
#if KCH_PQXX_MAJOR_VERSION < 7
            const bytestr_t mac_raw{row.at("l_mac")};
#else
            const auto mac_raw = row.at("l_mac").as<bytestr_t>();
#endif
            return {AdminDBClient::HostUpdate::Type::REMOVE,
                    isc::asiolink::IOAddress{row.at("l_ip").as<uint32_t>()},
                    isc::dhcp::HWAddr{util::to_bytea(mac_raw), isc::dhcp::HTYPE_ETHER},
                    row.at("l_subnet").as<uint32_t>()};
        }
 
        if (row.at("l_id").is_null()) {
#if KCH_PQXX_MAJOR_VERSION < 7
            const bytestr_t mac_raw{row.at("r_mac")};
#else
            const auto mac_raw = row.at("r_mac").as<bytestr_t>();
#endif
            return {AdminDBClient::HostUpdate::Type::ADD,
                    isc::asiolink::IOAddress{row.at("r_ip").as<uint32_t>()},
                    isc::dhcp::HWAddr{util::to_bytea(mac_raw), isc::dhcp::HTYPE_ETHER},
                    row.at("r_subnet").as<uint32_t>()};
        }
 
        throw std::runtime_error(__FILE__ ": compare_full_sync(): lines in JOIN without NULL id");
    }
 
    static AdminDBClient::HostUpdate make_host_update_from_hosts_row(const pqxx::row& row)
    {
#if KCH_PQXX_MAJOR_VERSION < 7
        const bytestr_t mac_raw{row.at("dhcp_identifier")};
#else
        const auto mac_raw = row.at("dhcp_identifier").as<bytestr_t>();
#endif
        const auto mac = util::to_bytea(mac_raw);
        const auto ip_raw = row.at("ipv4_address").as<uint32_t>();
        const auto subnet_id = row.at("dhcp4_subnet_id").as<uint32_t>();
 
        return {AdminDBClient::HostUpdate::Type::ADD, isc::asiolink::IOAddress{ip_raw},
                isc::dhcp::HWAddr{mac, isc::dhcp::HTYPE_ETHER}, subnet_id};
    }
 
    static void remove_tmp_host_table(pqxx::work& transaction)
    {
        transaction.exec0("DROP TABLE tmp_full_host_reservations;");
    }
 
    pqxx::connection psql;
};
 
LocalDBClient::LocalDBClient(std::string_view host, std::string_view port,
                             std::string_view username, std::string_view password,
                             std::string_view database)
{
    std::ostringstream connstr;
    connstr << "postgresql://" << username << ':' << password << '@' << host << ':' << port << '/'
            << database;
    p_impl = std::make_unique<LocalDBClientPrivate>(connstr.str());
}
 
LocalDBClient::~LocalDBClient() = default;
 
void LocalDBClient::apply_host_update(const AdminDBClient::HostUpdate& host_update)
{
    pqxx::work transaction{p_impl->psql};
    p_impl->apply_host_update(transaction, host_update);
    transaction.commit();
}
 
void LocalDBClient::apply_host_updates(const std::vector<AdminDBClient::HostUpdate>& host_updates)
{
    pqxx::work transaction{p_impl->psql};
    for (const auto& u : host_updates) {
        LOG_INFO(ahri::globals::logger, ahri::AHRI_HOST_UPDATE_INFO)
          .arg(u.mac.toText(false))
          .arg(u.ipv4.toText())
          .arg(u.incremental_update_id)
          .arg(u.subnet_id)
          .arg((u.type == ahri::AdminDBClient::HostUpdate::Type::ADD) ? "ADD" : "REMOVE");
        p_impl->apply_host_update(transaction, u);
    }
    transaction.commit();
}
 
std::vector<AdminDBClient::HostUpdate>
LocalDBClient::compare_full_sync(const std::vector<AdminDBClient::HostUpdate>& full_sync)
{
    p_impl->setup_tmp_host_table();
    p_impl->prepare_insert_tmp_host_reservations();
    pqxx::work transaction{p_impl->psql};
    p_impl->insert_tmp_host_reservations(transaction, full_sync);
 
    const pqxx::result differing_rows =
      p_impl->compare_current_and_tmp_host_reservations(transaction);
    std::vector<AdminDBClient::HostUpdate> diff;
    diff.reserve(differing_rows.size());
    for (const pqxx::row& row : differing_rows) {
        diff.push_back(LocalDBClientPrivate::make_host_update_from_diff_row(row));
    }
 
    LocalDBClientPrivate::remove_tmp_host_table(transaction);
    transaction.commit();
    return diff;
}
 
std::vector<AdminDBClient::HostUpdate>
LocalDBClient::query_host_reservation(const isc::dhcp::HWAddr& mac)
{
    pqxx::work txn{p_impl->psql};
 
#if KCH_PQXX_MAJOR_VERSION < 7
    const bytestr_t mac_bin = util::to_binstr(mac.hwaddr_);
#else
    const bytestr_t mac_bin = util::to_bytestr(mac.hwaddr_);
#endif
 
    const pqxx::result res = txn.exec_prepared(STMT_QUERY_HOST_RESERVATION_BY_MAC, mac_bin);
    std::vector<AdminDBClient::HostUpdate> hosts;
 
    for (const pqxx::row& row : res) {
        hosts.push_back(p_impl->make_host_update_from_hosts_row(row));
    }
 
    return hosts;
}
 
std::vector<AdminDBClient::HostUpdate>
LocalDBClient::query_host_reservation([[maybe_unused]] const isc::asiolink::IOAddress& ip_addr)
{
    pqxx::work txn{p_impl->psql};
 
    const pqxx::result res =
      txn.exec_prepared(STMT_QUERY_HOST_RESERVATION_BY_IP, ip_addr.toUint32());
    std::vector<AdminDBClient::HostUpdate> hosts;
 
    for (const pqxx::row& row : res) {
        hosts.push_back(LocalDBClientPrivate::make_host_update_from_hosts_row(row));
    }
 
    return hosts;
}
} // namespace ahri