measuring packet loss with C++

Measuring packet loss with Zabbix External Check and C++ Boost.Asio

Start

Zabbix is ​​good for everything, except for small timeouts for built-in and external checks. The server configuration file specifies a maximum timeout of 30 seconds, which is completely insufficient for, for example, a ping using 10,000 packets, even if we set the interval to 100ms. As we know, according to Russian standards, packet loss in communication networks should not exceed one packet per thousand. Particularly demanding customers interpret this figure in their own way: everything that is greater than or equal to 0.05% can be rounded up to the same 0.1% or one package per thousand. Therefore, I decided for the most critical nodes, especially when a request is received, to use an external program (let’s call it losshd), which will measure packet losses in several threads and write the results to the database, and use a simple utility for Zabbix (let’s call it getloss), which will quickly pull out the required value from the database.

Let’s describe the algorithm of our utility:

1. On request
   getloss -hlocalhost --dbname=zabbix --dbuser=zabbix --dbpass=mypassword --address=192.168.0.1
   the utility checks if the address 192.168.0.1 is in the table for the losshd daemon. If yes, it retrieves the value of the loss field from the database and prints it to the standard output. If not, it adds it to the table and displays the value 0 (we will assume that there are no losses until their presence is proven);

2. Next, we update the counter in the table last_read.

And here is the algorithm of the demon losshd:

1. We read from the database all ip-addresses that require verification;

2. We create separate flows for sending ICMP Echo Request packets for each address (hereinafter – senders);

3. We create one common stream for catching responses – ICMP Echo Reply (hereinafter – receiver);

4. We are waiting for the completion of all threads;

5. We consider the percentage of losses for each IP;

6. Write updates to the database;

7. After that, we remove from the table all rows that have last_read Last updated over a week ago. So we will reduce the load on the demon losshdremoving unclaimed checks;

8. We start everything from the beginning.

So, let’s get down to implementation. I will not post all the code here, otherwise the article will turn out to be too voluminous. You can see it on my github: https://github.com/Lordgprs/losshd.

First, we create a table for external validation. For simplicity, let’s make it in the same database that Zabbix server works with. I am using PostgreSQL:

CREATE TABLE ext_packetlosshd_dbg (
  ip inet NOT NULL PRIMARY KEY,
  loss DOUBLE PRECISION NOT NULL DEFAULT 0,
  last_update TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT NOW(),
  last_read TIMESTAMP WITHOUT TIME ZONE NOT NULL DEFAULT NOW()
);

Our C++ code will consist of six files: options.h, options.cpp, getloss.h, getloss.cpp, losshd.h, losshd.cpp. I will selectively provide declarations and definitions of classes and methods so as not to overload the article.

getloss utility

We write a base class for working with command line parameters:

class Options {
public:
  Options() = delete;
  Options(const Options &) = delete;
  Options(const Options &&) = delete;
  Options & operator=(const Options &) = delete;
  Options & operator=(const Options &&) = delete;
  Options (int, char **);
protected:
  po::options_description desc_;
  po::variables_map vm_;
};

Options::Options(int argc, char **argv):
desc_("Allowed options") { }

All default constructors are removed here, we specify only our constructor from parameters argc And argv (however, also empty). We plan to use the list of options and their values ​​in descendant classes, so we make them protected.

Writing an option recognition class for a utility getloss:

class OptionsGetloss : public Options {
public:
  OptionsGetloss() = delete;
  OptionsGetloss(const OptionsGetloss &) = delete;
  OptionsGetloss(const OptionsGetloss &&) = delete;
  OptionsGetloss & operator=(const OptionsGetloss &) = delete;
  OptionsGetloss & operator=(const OptionsGetloss &&) = delete;
  OptionsGetloss(int argc, char **argv);
  void CheckOptions();
  std::string get_dbname() const;
  std::string get_dbhost() const;
  std::string get_dbuser() const;
  std::string get_dbpass() const;
  std::string get_address() const;
};

Constructor and method that checks arguments:

OptionsGetloss::OptionsGetloss(int argc, char *argv[]) :
Options(argc, argv) {
  desc_.add_options()
    ("help", "this help")
    ("dbname,n", po::value<std::string>(), "database name")
    ("dbhost,h", po::value<std::string>(), "database server address")
    ("dbuser,U", po::value<std::string>(), "database user name")
    ("dbpass,P", po::value<std::string>(), "database password")
    ("address,A", po::value<std::string>(), "address to ping");
  po::store(po::parse_command_line(argc, argv, desc_), vm_);
  po::notify(vm_);

  CheckOptions();
}

void OptionsGetloss::CheckOptions() {
  if (vm_.count("help") != 0) {
    std::cout << desc_ << std::endl;
    std::exit(EXIT_SUCCESS);
  }
  if (
    vm_.count("dbname") == 0 ||
    vm_.count("dbhost") == 0 ||
    vm_.count("dbuser") == 0 ||
    vm_.count("address") == 0
   ) {
    std::cerr << "Error via processing arguments!" << std::endl << std::endl;
    std::cerr << desc_ << std::endl;
    std::exit(EXIT_FAILURE);
  }
}

Class for working with DBMS:

class Database {
public:
  Database() = delete;
  Database(const OptionsGetloss &);
  ~Database();
  double get_loss(const std::string &) const;
private:
  pqxx::connection conn_;
  pqxx::work mutable txn_;
};

double Database::get_loss(const std::string &ip) const {
  double loss = 0;
  std::string req = "SELECT loss FROM ext_packetlosshd_dbg WHERE ip = '" + ip + "'";
  auto result = txn_.exec(req);
  if (result.size() > 0) {
    loss = result[0][0].as<double>();
    txn_.exec("UPDATE ext_packetlosshd_dbg SET last_read = NOW() WHERE ip = '" + ip + "'");
  }
  else
    txn_.exec("INSERT INTO ext_packetlosshd_dbg (ip) VALUES ('" + ip + "')");
  return loss;
}

In function main() we create objects for parsing options and connecting to the DBMS and get a lossy figure from the database according to the algorithm described above:

int main(int argc, char **argv) {
  OptionsGetloss options(argc, argv);
  Database db(options);
  std::cout << db.get_loss(options.get_address()) << std::endl;
  return EXIT_SUCCESS;
}

We print the received value to the standard output so that the Zabbix server can read it from there. The line feed is left here for debugging – there is no need for it.

losshd daemon

Now let’s get to the heart of our test: the demon losshd. Description of structures Ipv4 And Icmpas well as basic methods, such as checksum calculation, are taken from the documentation for Boost.Asio almost unchanged:

class Ipv4 {
public:
  Ipv4();
  char8_t version() const;
  uint16_t header_length() const;
  char8_t tos() const;
  uint32_t time_to_live() const;
  boost::asio::ip::address_v4 source_address() const;
  uint32_t source_address_uint32() const;
  friend std::istream &operator>>(std::istream &is, Ipv4 &header);

private:
  char8_t data_[60];
};

class Icmp {
public:
  enum {
    kEchoReply = 0,
    kDestinationUnreachable = 3,
    kSourceQuench = 4,
    kRedirect = 5,
    kEchoRequest = 8,
    kTimeExceeded = 11,
    kParameterProblem = 12,
    kTimestampRequest = 13,
    kTimestampReply = 14,
    kInfoRequest = 15,
    kInfoReply = 16,
    kAddressRequest = 17,
    kAddressReply = 18
  };
  Icmp();
  char8_t type() const;
  char8_t code() const;
  uint16_t identifier() const;
  uint16_t sequence_number() const;
  void type(char8_t);
  void code(char8_t);
  void checksum(uint16_t);
  void identifier(uint16_t);
  void sequence_number(uint16_t);
  void CalculateChecksum(auto body_begin, auto body_end);
  friend std::istream& operator>>
    (std::istream &inputStream, Icmp &header);
  friend std::ostream& operator<<
    (std::ostream &outputStream, const Icmp &header);

private:
  uint16_t Decode(int32_t a, int32_t b) const;
  void Encode(int32_t a, int32_t b, uint16_t n);

  char8_t data_[8];
};

Ipv4::Ipv4() {
  std::fill (data_, data_ + sizeof(data_), 0);
}

char8_t Ipv4::version() const {
  return (data_[0] >> 4) & 0xF;
}

uint16_t Ipv4::header_length() const {
  return (data_[0] & 0xF) << 2;
}

char8_t Ipv4::tos() const {
  return data_[1];
}

uint32_t Ipv4::time_to_live() const {
  return data_[8];
}

boost::asio::ip::address_v4 Ipv4::source_address() const {
  boost::asio::ip::address_v4::bytes_type bytes = {
    {data_[12], data_[13], data_[14], data_[15]}
  };
  return boost::asio::ip::address_v4(bytes);
}

uint32_t Ipv4::source_address_uint32() const {
  return (data_[12] << 24) | 
    (data_[13] << 16) | 
    (data_[14] << 8) |
    data_[15];
}

Icmp::Icmp() {
  std::fill(data_, data_ + sizeof(data_), 0);
}

char8_t Icmp::type() const {
  return data_[0];
}

char8_t Icmp::code() const {
  return data_[1];
}

uint16_t Icmp::identifier() const {
  return Decode(4, 5);
}

uint16_t Icmp::sequence_number() const {
  return Decode(6, 7);
}

void Icmp::type(char8_t n) {
  data_[0] = n;
}

void Icmp::code(char8_t n) {
  data_[1] = n;
}

void Icmp::checksum(uint16_t n) {
  Encode(2, 3, n);
}

void Icmp::identifier(uint16_t n) {
  Encode(4, 5, n);
}

void Icmp::sequence_number(uint16_t n) {
  Encode(6, 7, n);
}

void Icmp::CalculateChecksum(auto body_begin, auto body_end) {
  uint32_t sum = (type() << 8) + code() + identifier() + sequence_number();
  auto body_iterator = body_begin;
  while (body_iterator != body_end) {
    sum += (static_cast<char8_t>(*body_iterator++) << 8);
    if (body_iterator != body_end)
      sum += static_cast<char8_t>(*body_iterator++);
  }

  sum = (sum >> 16) + (sum & 0xFFFF);
  sum += (sum >> 16);
  checksum(static_cast<uint16_t>(~sum));
}

uint16_t Icmp::Decode(const int32_t a, const int32_t b) const {
  return (data_[a] << 8) + data_[b];
}

void Icmp::Encode(const int32_t a, const int32_t b, const uint16_t n) {
  data_[a] = static_cast<char8_t>(n >> 8);
  data_[b] = static_cast<char8_t>(n & 0xFF);
}

std::istream& operator>>(std::istream& input_stream, Icmp &header) {
  return input_stream.read(reinterpret_cast<char *>(header.data_), 8);
}

std::ostream& operator<<(std::ostream& output_stream, const Icmp &header) {
  return output_stream.write(
    reinterpret_cast<const char *>(header.data_), 8);
}

std::istream &operator>>(std::istream &is, Ipv4 &header) {
  is.read(reinterpret_cast<char *>(header.data_), 20);
  if (header.version() != 4)
    is.setstate(std::ios::failbit);
  std::streamsize options_length = header.header_length() - 20;
  if (options_length < 0 || options_length > 40)
    is.setstate(std::ios::failbit);
  else
    is.read(reinterpret_cast<char *>(header.data_) + 20, options_length);
  return is;
}

Let’s create classes IcmpSender And IcmpReceiver for respectively sending and receiving icmp packets:

class IcmpSender {
public:
  IcmpSender() = delete;
  IcmpSender(const IcmpSender &) = delete;
  IcmpSender(const IcmpSender &&) = delete;
  IcmpSender & operator=(const IcmpSender &) = delete;
  IcmpSender & operator=(const IcmpSender &&) = delete;
  IcmpSender(boost::asio::io_context &, const char *, uint16_t, 
             uint16_t, std::mutex &, std::condition_variable &);

private:
  void StartSend();

  icmp::endpoint destination_;
  icmp::socket socket_;
  uint16_t sequence_number_;
  uint16_t count_;
  chrono::steady_clock::time_point time_sent_;
  boost::asio::streambuf reply_buffer_;
  std::size_t interval_;
  std::size_t size_;
  std::mutex &mtx_;
  std::condition_variable &condition_;
};

class IcmpReceiver {
public:
  IcmpReceiver() = delete;
  IcmpReceiver(const IcmpReceiver &) = delete;
  IcmpReceiver(const IcmpReceiver &&) = delete;
  IcmpReceiver & operator=(const IcmpReceiver &) = delete;
  IcmpReceiver & operator=(const IcmpReceiver &&) = delete;
  IcmpReceiver(boost::asio::io_context &, int &, 
               std::unordered_map<uint32_t, uint32_t> &, 
               std::mutex &, std::condition_variable &);

private:
  void StartReceive();
  void CheckIfSendersExist();
  void HandleReceive(std::size_t length);

  icmp::socket socket_;
  boost::asio::streambuf reply_buffer_;
  int &senders_count_;
  std::unordered_map<uint32_t, uint32_t> &ping_results_;
  std::mutex &mtx_;
  std::condition_variable &condition_;
  bool senders_unlocked_ = false;
  boost::asio::deadline_timer dt_;
  const boost::posix_time::time_duration 
    kReceiveTimerFrequency = boost::posix_time::seconds(5);
};

Let’s go through some fields of the class IcmpSender:

• destination_ – IP that we will ping;

• socket_ – Asio socket for sending ICMP packets;

• sequence_number_ – ICMP packet number;

• count_, interval_, size_ – respectively, the number of sent packets, the interval between sendings and the packet size;

• mtx_ And condition_variable_ – references to a mutex and a state variable for thread synchronization.

In class IcmpReceiver we will have the following fields:

• socket socket_ and buffer reply_buffer_ to receive an ICMP response;

• senders_count_ – a reference to a variable containing the number of active sender threads. When this field is zero, we will stop catching responses;

• ping_results_ – the results of our ping (the number of received packets). As a key, we use the IP address translated into a 32-bit number;

• Just like in the sender class, here we use references to a shared mutex and a state variable;

• kReceiveTimerFrequency – with this constant we set the number of seconds after which the timer will check if there are active sender threads.

Consider the process of sending packets:

void IcmpSender::StartSend() {
  std::string body(size_ - sizeof(Icmp), '\0');
  // Create an ICMP header
  Icmp icmp;
  icmp.type(Icmp::kEchoRequest);
  icmp.code(0);
  icmp.identifier(static_cast<uint16_t>(getpid()));
  for (size_t i = 0; i < count_; i++) {
    icmp.sequence_number(++sequence_number_);
    icmp.CalculateChecksum(body.begin(), body.end());
    // Encode the request packet
    boost::asio::streambuf requestBuffer;
    std::ostream outputStream(&requestBuffer);
    outputStream << icmp << body;

    // Send the request
    socket_.send_to(requestBuffer.data(), destination_);
    std::this_thread::sleep_for(std::chrono::milliseconds(interval_));
  }
}

Everything is simple here: we fill the object with data and send it to the socket using the overloaded shift operator. Between iterations, we simply fall asleep on the given field interval_ number of milliseconds.

Now consider the process of catching responses. Here everything is more complicated:

void IcmpReceiver::StartReceive() {
  // Discard any data already in the buffer
  reply_buffer_.consume(reply_buffer_.size());
  if ((!senders_unlocked_) || (senders_unlocked_ && senders_count_ > 0)) {
    socket_.async_receive(reply_buffer_.prepare(65536),
      boost::bind(&IcmpReceiver::HandleReceive, this, boost::placeholders::_2));
    dt_.async_wait(boost::bind(&IcmpReceiver::CheckIfSendersExist, this));
  }
  if (!senders_unlocked_) {
    {
      std::unique_lock<std::mutex> ul(mtx_);
      // Notifying senders: we are ready to catch echo replies
      condition_.notify_all();
    }
    senders_unlocked_ = true;
  }
}

Binding a socket to a method HandleReceive, which will process incoming packets. We initialize a timer that checks if the sender threads are alive (CheckIfSendersExists). We are ready to receive packets – we inform other threads using the method notify_all() state variable. When the sender threads stop their work, we stop the timer and the socket:

void IcmpReceiver::CheckIfSendersExist() {
  int count;
  {
    std::unique_lock<std::mutex> ul(mtx_);
    count = senders_count_;
  }
  // Stopping catching replies if all senders have been finished
  if (count == 0) {
    dt_.cancel();
    socket_.cancel();
  }
}

While the sending is in progress, we catch the packets:

void IcmpReceiver::HandleReceive(std::size_t length) {
  reply_buffer_.commit(length);

  // Decode the reply packet
  std::istream input_stream(&reply_buffer_);
  Ipv4 ipv4;
  Icmp icmp;
  input_stream >> ipv4 >> icmp;
  if (input_stream && icmp.type() == Icmp::kEchoReply
  && icmp.identifier() == static_cast<uint16_t>(getpid())) {
    dt_.cancel();
    uint32_t src = ipv4.source_address_uint32();
    {
      std::lock_guard lg(mtx_);
      ping_results_[src]++;
    }
  }
  {
    std::lock_guard lg(mtx_);
    if(senders_count_ > 0)
      StartReceive();
  }
}

We increase the number of received packets for the IP from which the response came in the container ping_results_.

So, we figured out how to send and receive packets. Let’s write a scheduler class that will be responsible for initializing threads and writing the result:

class Scheduler {
public:
  Scheduler(OptionsLosshd &);
  ~Scheduler();
  void Run();
  void Clean();

private:
  uint32_t GetIpFromString (const std::string &str_ip);
  std::string GetIpFromUint32 (const uint32_t ip) const;
  std::vector<std::string> GetAddressesForPing() const;
  auto CreateReceiver();
  auto CreateSender();

  OptionsLosshd &options_;
  pqxx::connection conn_;
  pqxx::nontransaction mutable txn_;
  std::mutex mtx_;
  std::vector<std::string> address_list_;
  std::unordered_map<uint32_t, uint32_t> ping_results_;
  std::condition_variable condition_;
};

class constructor Scheduler we only pass a reference to the command line options. Methods CreateReceiver And CreateSender created to increase the readability of the code – they return a lambda expression that is passed to the thread’s constructor std::thread. Our scheduler has the following private fields:

• options_ – set of command line options;

• conn_ – connection with the DBMS;

• txn_ – processing queries to the database;

• address_list_ – a list of addresses that we will ping;

• ping_results_ – already familiar container with ping results;

• condition_, mtx_ – thread synchronization.

We get a list of addresses that we will ping, simultaneously deleting from the table the addresses for which statistics have not been taken for more than a week:

std::vector<std::string> Scheduler::GetAddressesForPing() const {
  std::string req = "";
  std::vector<std::string> addresses;
  txn_.exec("DELETE FROM ext_packetlosshd_dbg WHERE last_read + interval '7 days' < NOW()");
  for (auto row: txn_.exec("SELECT ip FROM ext_packetlosshd_dbg LIMIT 100"))
    addresses.push_back(row[0].c_str());
  return addresses;
}

Starting threads:

void Scheduler::Run() {
  std::vector<std::thread> threads;
  int senders_count = address_list_.size();
  // Starting ICMP receiver
  std::thread t(CreateReceiver(), std::ref(senders_count), 
                std::ref(ping_results_), std::ref(mtx_), std::ref(condition_));
  // Starting ICMP senders
  for (size_t i = 0; i < address_list_.size(); i++) {
    std::thread t(CreateSender(), address_list_[i], 
                  std::ref(senders_count), std::ref(mtx_), 
                  std::ref(condition_), std::ref(options_));
    threads.push_back(std::move
  }
  // Joining senders
  for (size_t i = 0; i < threads.size(); i++)
     threads.at(i).join();
  // Joining receiver
  t.join();
  std::cout << "End of collecting results:" << std::endl;
  for (auto i: ping_results_) {
    std::cout << "from IP " <<  GetIpFromUint32(i.first) << 
      " received: " << i.second << std::endl;
    txn_.exec("UPDATE ext_packetlosshd_dbg SET \
        loss = " + std::to_string(
          static_cast<double>(options_.get_count() - i.second) / 
          options_.get_count() * 100) + ", " + " \
        last_update = NOW() \
        WHERE \
        ip = '" + GetIpFromUint32(i.first) + "'");
  }
  std::cout << std::endl;
}

After completing their work, we add the results to the database.

Separately, I want to consider the flow-sender:

auto Scheduler::CreateSender() {
  return [](std::string address, int &senders, std::mutex &mtx, 
            std::condition_variable &condition, OptionsLosshd &options) {
    {
      std::unique_lock<std::mutex> ul(mtx);
      // Waiting for receiver starts
      condition.wait(ul);
    }
    boost::asio::io_context io_context;
    IcmpSender pinger(io_context, address.c_str(), options.get_count(), 
                      options.get_interval(), mtx, condition);
    {
      std::lock_guard lg(mtx);
      senders--;
    }
  };
}

All sender threads do not start sending packets until they receive a notification from the receiver thread using the method notify_all().

After completing the collection of statistics, you need to prepare for a new iteration. To do this, we again obtain the current list of addresses for monitoring and clear the container with the results:

void Scheduler::Clean() {
  ping_results_.clear();
  address_list_ = GetAddressesForPing();
  for (auto address: address_list_)
    ping_results_.insert({GetIpFromString(address), 0});
}

Well, let’s finally define the function main():

int main(int argc, char *argv[]) {
  constexpr auto kPauseBetweenIterations = std::chrono::seconds(5);
  OptionsLosshd options(argc, argv);
  if (options.is_daemon()) {
    std::cout << "Running as a daemon..." << std::endl;
    auto pid = fork();
    if (pid > 0)
      return EXIT_SUCCESS;
    if (pid < 0) {
      std::cerr << "Error while doing fork()! Exiting...";
      return EXIT_FAILURE;
    }
    umask(0);
    setsid();
    if (chdir("/") < 0) {
      std::cerr << "Error while attempting chdir()!" << std::endl;
      return EXIT_FAILURE;
    };
    close(STDIN_FILENO);
    close(STDOUT_FILENO);
    close(STDERR_FILENO);
  }
  Scheduler scheduler(options);
  while (true) {
    std::cout << "New iteration started. Pinging hosts..." << std::endl;
    scheduler.Run();
    scheduler.Clean();
    std::this_thread::sleep_for(kPauseBetweenIterations);
  }
  return EXIT_SUCCESS;
}

We implement the work of the daemon in the classic UNIX way: using the system call fork().

Assembly and launch

We collect and run our project. I will be doing this on Ubuntu 22.04 LTS using CMake and conan:

[requires]
boost/1.74.0
libpqxx/6.4.8

[generators]
CMakeDeps
CMakeToolchain

cmake_policy(SET CMP0048 NEW)
project(LossHD VERSION 1.0.3)
cmake_minimum_required(VERSION 3.18)
set(CMAKE_CXX_FLAGS "-O2")
find_package(Boost REQUIRED COMPONENTS program_options REQUIRED)
find_package(libpqxx REQUIRED)
add_executable(losshd src/losshd.cpp src/options.cpp)
set_property(TARGET losshd PROPERTY CXX_STANDARD 20)
set_property(TARGET losshd PROPERTY CXX_STANDARD_REQUIRED On)
set_property(TARGET losshd PROPERTY CXX_EXTENSIONS Off)
target_link_libraries(losshd boost_program_options pqxx)
add_executable(getloss src/getloss.cpp src/options.cpp)
set_property(TARGET getloss PROPERTY CXX_STANDARD 20)
set_property(TARGET getloss PROPERTY CXX_STANDARD_REQUIRED On)
set_property(TARGET getloss PROPERTY CXX_EXTENSIONS Off)
target_link_libraries(getloss boost_program_options pqxx)

mkdir build
cd build
conan install .. -of ../conan
cmake .. -DCMAKE_TOOLCHAIN_FILE=../conan/conan_toolchain.cmake \
  -DCMAKE_BUILD_TYPE=Release
cmake --build .

We got two executable files in the build directory: getloss and losshd. We launch losshd:

./losshd -nzabbix -h127.0.0.1 -Upostgres -Ppassword -i100 -s1400 -c10000 --daemon

Copy getloss to catalog /usr/local/bin. In the directory with external zabbix checks, create a bash script getloss.sh:

#!/bin/bash

/usr/local/bin/getloss -nzabbix -h127.0.0.1 -Upostgres -Ppassword -A $1

Setting up Zabbix

Create a new empty template. I named it ICMP Packet Loss HD. Create a new data element in the template:

We add a template to network nodes that need an accurate measurement of the quality of communication. We are waiting and enjoying such charts:

Conclusion

The service can already be used, but some things still need to be completed:

1. Implement logging to an arbitrary stream instead of std::coutand possibly to the base;

2. Add exception handling to the libpqxx library (breakdown of the connection to the DBMS, etc.).

Since I’m just starting to learn C++, I’ll be glad for any criticism that will make my code better. Thank you all for your attention, I hope my note helped you too.