Merge pull request #379 from CopernicaMarketingSoftware/randomIps

Add option to select an IP randomly

src/linux_tcp/address2str.h

@@ -0,0 +1,108 @@
+/**
+ *  Address2Str.h
+ * 
+ *  Helper class to get a stringed version of an address obtained from get_addrinfo
+ *  
+ *  @author Aljar Meesters <aljar.meesters@copernica.com
+ *  @copyright 2020 Copernica BV
+ */
+
+/**
+ *  Include guard
+ */
+#pragma once
+
+/**
+ *  Dependencies
+ */
+#include <sys/socket.h>
+#include <arpa/inet.h>
+
+/**
+ *  Opening namespace
+ */
+namespace AMQP {
+
+/**
+ *  class implemenation
+ */
+class Address2str
+{
+private:
+    /**
+     *  The address as string
+     */
+    char *_buffer = nullptr;
+
+public:
+    /**
+     *  Constructor
+     *  @param struct addrinfo *
+     */
+    Address2str(struct addrinfo *info)
+    {
+        // Switch on family
+        switch(info->ai_family)
+        {
+            // If we are ip4
+            case AF_INET: 
+            {
+                // ugly cast
+                struct sockaddr_in *addr_in = (struct sockaddr_in *)(info->ai_addr);
+                
+                // create the buffer
+                _buffer = new char[INET_ADDRSTRLEN];
+                
+                // get the info in our buffer
+                inet_ntop(AF_INET, &(addr_in->sin_addr), _buffer, INET_ADDRSTRLEN);
+                
+                // done
+                break;
+            }
+
+            // if we are ipv6
+            case AF_INET6:
+            {
+                // ugly cast
+                struct sockaddr_in6 *addr_in6 = (struct sockaddr_in6 *)(info->ai_addr);
+
+                // crate buffer
+                _buffer = new char[INET6_ADDRSTRLEN];
+
+                // get the info in our buffer
+                inet_ntop(AF_INET6, &(addr_in6->sin6_addr), _buffer, INET6_ADDRSTRLEN);
+                
+                // done
+                break;
+            }
+            default:
+                // If it is not ipv4 or ipv6 we don't set it.
+                break;
+        }
+    }
+
+    /**
+     *  Destruction
+     */
+    virtual ~Address2str()
+    {
+        // if we have a buffer, we should free it.
+        if (_buffer) delete[](_buffer);
+    }
+
+    /**
+     *  get the char
+     *  @return char*
+     */
+    const char *toChar() const
+    {
+        // expose the buffer.
+        return _buffer;
+    }
+};
+
+
+/**
+ * ending of namespace
+ */
+}

src/linux_tcp/addressinfo.h

@@ -1,12 +1,19 @@
 /**
  *  AddressInfo.h
  *
- *  Utility wrapper arround "getAddressInfo()"
+ *  Utility wrapper around "getAddressInfo()"
  *
  *  @author Emiel Bruijntjes <emiel.bruijntjes@copernica.com>
  *  @copyright 2015 Copernica BV
  */
 
+/**
+ *  Dependencies
+ */
+#include <random>
+#include "connectionorder.h"
+#include "address2str.h"
+
 /**
  *  Include guard
  */
@@ -30,13 +37,103 @@ private:
      */
     std::vector<struct addrinfo *> _v;
 
+    /**
+     *  Helper function to order the vector of addrinfo based on the ordering received
+     *  @param order
+     */
+    void reorder(ConnectionOrder::Order order)
+    {    
+        // witch on order
+        switch (order)
+        {
+            // Do we want to have a random order of the addresses?
+            // This may be useful since getaddrinfo is sorting the addresses on proximity
+            // (e.g. https://lists.debian.org/debian-glibc/2007/09/msg00347.html),
+            // which may break loadbalancing..
+            case ConnectionOrder::Order::random:
+            {
+                // create a random device for the seed of the random number generator
+                std::random_device rd;
+
+                // Create the generator
+                std::mt19937 gen(rd());
+
+                // shuffle the vector.
+                std::shuffle(_v.begin(), _v.end(), gen);
+
+                // done
+                break;
+            }
+            // do we want to sort in ascending order
+            case ConnectionOrder::Order::ascending:
+            {
+                std::sort(_v.begin(), _v.end(), []
+                (struct addrinfo * v1, struct addrinfo * v2) -> bool 
+                {
+                    // get the addresses
+                    Address2str addr1(v1);
+                    Address2str addr2(v2);
+                    
+                    // if addr1 doesn't have a proper address it should go to the
+                    // back. Same holds for addr2
+                    if (addr1.toChar() == nullptr) return false;
+                    if (addr2.toChar() == nullptr) return true;
+
+                    // make the comparison based on string comparison
+                    return strcmp(addr1.toChar(), addr2.toChar()) < 0;
+                });
+
+                // done
+                break;
+            }
+
+            // do we want to sort in descending order
+            case ConnectionOrder::Order::descending:
+            {
+                std::sort(_v.begin(), _v.end(), []
+                (struct addrinfo * v1, struct addrinfo * v2) -> bool
+                {
+                    // get the addresses
+                    Address2str addr1(v1);
+                    Address2str addr2(v2);
+                    
+                    // if addr1 doesn't have a proper address it should go to the
+                    // back. Same holds for addr2
+                    if (addr1.toChar() == nullptr) return false;
+                    if (addr2.toChar() == nullptr) return true;
+
+                    // make the comparison based on string comparison
+                    return strcmp(addr1.toChar(), addr2.toChar()) > 0;
+                });
+
+                // done
+                break;
+            }
+
+            // de we want to have reverse ordering of proximity
+            case ConnectionOrder::Order::reverse:
+            { 
+                std::reverse(_v.begin(), _v.end());
+
+                // done
+                break;
+            }
+
+            default:
+                // nothing to do, just default behaviour
+                break;
+        }
+    }
+
+
 public:
     /**
      *  Constructor
      *  @param  hostname
      *  @param  port
+     *  @param  order
      */
-    AddressInfo(const char *hostname, uint16_t port = 5672)
+    AddressInfo(const char *hostname, uint16_t port = 5672, ConnectionOrder::Order order = ConnectionOrder::Order::standard)
     {
         // store portnumber in buffer
         auto portnumber = std::to_string(port);
@@ -63,6 +160,9 @@ public:
             // store in vector
             _v.push_back(current);
         }
+
+        // Order the vector based on the provided ordering
+        reorder(order);
     }
 
     /**
@@ -99,4 +199,3 @@ public:
  *  End of namespace
  */
 }
-

src/linux_tcp/connectionorder.h

@@ -0,0 +1,73 @@
+/**
+ *  ConnectionOrder.h
+ *  
+ *  Class that give info on how we want to sellect the connection from a list of IPs
+ * 
+ *  @author Aljar Meesters <aljar.meesters@copernica.com>
+ *  @copyright 2020 Copernica BV
+ */
+
+/**
+ *  Include guard
+ */
+#pragma once
+
+/**
+ *  Setup namespace
+ */
+namespace AMQP
+{
+
+/**
+ *  Class implementation
+ */
+class ConnectionOrder
+{
+public:
+    /**
+     *  Enum class holding the orders we support
+     *  - standard:   what is used by getaddrinfo, which is proximity based
+     *  - reverse:    reverse the standard order
+     *  - random:     random order
+     *  - ascending:  try the smallest IP address first
+     *  - descending: try the largest IP address first
+     *  @var enum Order
+     */
+    enum class Order { standard, reverse, random, ascending, descending};
+
+private: 
+    /**
+     *  The order for the connection
+     *  @var Order
+     */
+    Order _order;
+
+public:
+    /**
+     *  Constructor
+     *  @var  order
+     */
+    ConnectionOrder (const char *order) : _order(Order::standard)
+    {
+        // Set the orders based on the string
+        if (strcmp(order, "random") == 0) _order = Order::random;
+        else if (strcmp(order, "ascending") == 0 || strcmp(order, "asc") == 0) _order = Order::ascending;
+        else if (strcmp(order, "descending") == 0 || strcmp(order, "desc") == 0 ) _order = Order::descending;
+
+        // If we don't ave a match we fall back to standard, which was set as default
+    }
+
+    /**
+     *  Get the order
+     *  @return Order
+     */
+    Order order()
+    {
+        return _order;
+    }
+};
+
+/**
+ *  End of namespace
+ */
+}

src/linux_tcp/tcpconnection.cpp

@@ -26,7 +26,7 @@ namespace AMQP {
  */
 TcpConnection::TcpConnection(TcpHandler *handler, const Address &address) :
     _handler(handler),
-    _state(new TcpResolver(this, address.hostname(), address.port(), address.secure(), address.option("connectTimeout", 5))),
+    _state(new TcpResolver(this, address.hostname(), address.port(), address.secure(), address.option("connectTimeout", 5), ConnectionOrder(address.option("connectionOrder")).order())),
     _connection(this, address.login(), address.vhost()) 
 {
     // tell the handler
@@ -286,4 +286,3 @@ void TcpConnection::onLost(TcpState *state)
  *  End of namespace
  */
 }
-

src/linux_tcp/tcpresolver.h

@@ -85,6 +85,12 @@ private:
      */
     std::thread _thread;
 
+    /**
+     *  How should the addresses be ordered when we want to connect
+     *  @var bool
+     */
+    ConnectionOrder::Order _order;
+
 
     /**
      *  Run the thread
@@ -98,7 +104,7 @@ private:
             if (_secure && !OpenSSL::valid()) throw std::runtime_error("Secure connection cannot be established: libssl.so cannot be loaded");
             
             // get address info
-            AddressInfo addresses(_hostname.data(), _port);
+            AddressInfo addresses(_hostname.data(), _port, _order);
     
             // the pollfd structure, needed for poll()
             pollfd fd;
@@ -188,13 +194,15 @@ public:
      *  @param  portnumber  The portnumber for the lookup
      *  @param  secure      Do we need a secure tls connection when ready?
      *  @param  timeout     timeout per connection attempt
+     *  @param  order       How should we oreder the addresses of the host to connect to
      */
-    TcpResolver(TcpParent *parent, std::string hostname, uint16_t port, bool secure, int timeout) : 
+    TcpResolver(TcpParent *parent, std::string hostname, uint16_t port, bool secure, int timeout, ConnectionOrder::Order order) : 
         TcpExtState(parent), 
         _hostname(std::move(hostname)),
         _secure(secure),
         _port(port),
-        _timeout(timeout)
+        _timeout(timeout),
+        _order(order)
     {
         // tell the event loop to monitor the filedescriptor of the pipe
         parent->onIdle(this, _pipe.in(), readable);