Exporter.cpp

Go to the documentation of this file.

#include <exception>
 
#include <xmlrpc-c/girerr.hpp>
 
#include <log/macros.h>
 
#include "Exporter.hpp"
#include "util/ArpwatchRPC.hpp"
#include "util/HiredisAdapter.hpp"
 
#include "log_admindb_arpwatch_exporter.h"
 
namespace
{
constexpr uint32_t SUBNET_ID_TO_VLAN_MODULUS{10000};
 
std::string get_ip2mac_key(const arpwatch::Exporter::ip4_addr_t& ip)
{
    return "ip2mac-" + ip.toText();
}
 
std::string get_mac2ip_key(const arpwatch::Exporter::mac_addr_t& mac)
{
    return "mac2ip-" + mac.toText(false);
}
 
std::string get_ip2macWithVLAN_key(const arpwatch::Exporter::ip4_addr_t& ip,
                                   const arpwatch::Exporter::subnet_id_t subnet_id)
{
    const uint16_t vlan = subnet_id % SUBNET_ID_TO_VLAN_MODULUS;
    return "ip2macWithVLAN-" + std::to_string(vlan) + '-' + ip.toText();
}
} // namespace
 
namespace arpwatch
{
Exporter::~Exporter()
{
    if (worker_keep_running) {
        stop_worker();
    }
}
 
void Exporter::start_worker()
{
    LOG_DEBUG(logger, 1, EXPORTER_REDIS_WORKER_START);
    worker_keep_running = true;
    worker = std::make_unique<std::thread>([this]() { run_worker(); });
}
 
void Exporter::stop_worker()
{
    LOG_DEBUG(logger, 1, EXPORTER_REDIS_WORKER_STOP);
    worker_keep_running = false;
    worker_notify_cv.notify_all();
    if (worker->joinable()) {
        worker->join();
    }
}
 
void Exporter::push_lease(const Exporter::mac_addr_t& mac, const Exporter::ip4_addr_t& ip4,
                          const Exporter::subnet_id_t subnet_id,
                          const std::chrono::seconds valid_lft)
{
    LOG_DEBUG(logger, 10, EXPORTER_WRITE_REDIS)
      .arg(mac.toText(false).c_str())
      .arg(ip4.toText().c_str());
    std::unique_lock<std::mutex> lck{lease_queue_mtx};
    lease_queue.emplace(mac, ip4, subnet_id, valid_lft);
    lck.unlock();
    worker_notify_cv.notify_all();
}
 
void Exporter::flush_lease_queue()
{
    LOG_DEBUG(logger, 20, EXPORTER_FLUSH_LEASE_QUEUE);
 
    std::unique_lock<std::mutex> lck{lease_queue_mtx};
    while (not lease_queue.empty()) {
        lck.unlock();
        worker_notify_cv.notify_all();
        std::this_thread::sleep_for(LEASE_QUEUE_FLUSH_WAIT);
        lck.lock();
    }
}
 
void Exporter::run_worker()
{
    while (worker_keep_running) {
        std::unique_lock<std::mutex> cv_lk{lease_queue_mtx};
        worker_notify_cv.wait(
          cv_lk, [this]() { return not lease_queue.empty() || not worker_keep_running; });
 
        if (!redis_adapter.is_connected()) {
            try {
                redis_adapter.connect();
            }
            catch (HiredisException& e) {
                LOG_WARN(logger, HIREDIS_ERROR).arg(e.what());
                cv_lk.unlock();
                std::this_thread::sleep_for(REDIS_RECONNECT_WAIT);
                continue;
            }
        }
 
        while (not lease_queue.empty()) {
            assert(cv_lk.owns_lock());
 
            const auto& lease = lease_queue.front();
            cv_lk.unlock();
 
            try {
                worker_send_lease_to_redis(lease);
                worker_send_lease_to_arpwatch(lease);
            }
            catch (HiredisException& ex) {
                LOG_WARN(logger, HIREDIS_ERROR).arg(ex.what());
                // Start from the top of the outer loop, thus forcing a reconnect attempt
                break;
            }
            catch (girerr::error& ex) {
                LOG_WARN(logger, EXPORTER_RPC_ERROR).arg(ex.what());
            }
            catch (std::exception& ex) {
                LOG_WARN(logger, EXPORTER_PUSH_ERROR).arg(ex.what());
            }
 
            cv_lk.lock();
            lease_queue.pop();
        }
    }
}
 
void Exporter::worker_send_lease_to_redis(const lease4_t& lease)
{
    const auto& [mac, ip, subnet_id, valid_lft] = lease;
    std::string ip2mac_key = get_ip2mac_key(ip);
    std::string mac2ip_key = get_mac2ip_key(mac);
    std::string ip2macWithVLAN_key = get_ip2macWithVLAN_key(ip, subnet_id);
    std::string mac_value = mac.toText(false);
    std::string ip4_value = ip.toText();
 
    const auto expire =
      static_cast<std::uint32_t>(static_cast<double>(valid_lft.count()) * REDIS_EXPIRE_FACT);
 
    LOG_DEBUG(logger, 10, EXPORTER_SEND_REDIS)
      .arg(ip2mac_key.c_str())
      .arg(mac_value.c_str())
      .arg(expire);
    redis_adapter.set(ip2mac_key, mac_value, expire);
 
    LOG_DEBUG(logger, 10, EXPORTER_SEND_REDIS)
      .arg(mac2ip_key.c_str())
      .arg(ip4_value.c_str())
      .arg(expire);
    redis_adapter.sadd(mac2ip_key, ip4_value, expire);
 
    if (subnet_id != 0) {
        LOG_DEBUG(logger, 10, EXPORTER_SEND_REDIS)
          .arg(ip2macWithVLAN_key)
          .arg(mac_value)
          .arg(expire);
        redis_adapter.set(ip2macWithVLAN_key, mac_value, expire);
    }
}
 
void Exporter::worker_send_lease_to_arpwatch(const lease4_t& lease)
{
    const auto& [mac, ip4, subnet_id, valid_lft] = lease;
    const auto vlan_id = subnet_id % SUBNET_ID_TO_VLAN_MODULUS;
    arpwatch_rpc.push_seen_entry(vlan_id, mac, ip4);
}
 
} // namespace arpwatch