add an ability to run simulation against loaded katran's bpf program (f…

…acebookincubator#41)

Summary:
i've added an ability to run simulation for specified 5 tuple: user would be able to provide 5 tuple (src/dst address + src/dst ports + protocol number) and would receive address of the real, where this 5 tuple is going to be sent. this could be used for debug/troubleshooting scenarios (e.g. some flows works slow/does not work at all and we want to figure out which real is responsible for serving em). all this simulation is being run against loaded BPF program (so if you have features, like src based routing enable, it would work for em out of box))

added dedicated section to katran_tester which implements integration tests against this new freature
Pull Request resolved: facebookincubator#41

Pulled By: udippant

Test Plan:
Imported from GitHub, without a `Test Plan:` line.
Unit test

Reviewed By: yangchi

Differential Revision: D17076371

fbshipit-source-id: c12affd44b914233f40c7595693117d0b6b03752

katran/lib/CMakeLists.txt

@@ -115,6 +115,15 @@ set(PCAPWRITER_DEPS
 
 target_link_libraries(pcapwriter "${PCAPWRITER_DEPS}")
 
+add_library(katransimulator STATIC
+  KatranSimulator.h
+  KatranSimulator.cpp
+)
+
+target_link_libraries(katransimulator
+    bpfadapter
+)
+
 add_library(katranlb STATIC
     KatranEventReader.h
     KatranEventReader.cpp
@@ -134,6 +143,7 @@ target_link_libraries(katranlb
     chhelpers
     iphelpers
     pcapwriter
+    katransimulator
     "${GFLAGS}"
     "${PTHREAD}"
     "-Wl,--end-group"

katran/lib/KatranLb.cpp

@@ -38,6 +38,7 @@ constexpr int kMaxForwardingCores = 128;
 constexpr int kFirstElem = 0;
 constexpr int kError = -1;
 constexpr uint32_t kMaxQuicId = 65535; // 2^16-1
+constexpr folly::StringPiece kEmptyString = "";
 } // namespace
 
 KatranLb::KatranLb(const KatranConfig& config)
@@ -1202,6 +1203,15 @@ std::unordered_map<uint32_t, std::string> KatranLb::getHealthcheckersDst() {
   return hcs;
 }
 
+const std::string KatranLb::getRealForFlow(const KatranFlow& flow) {
+  if (!progsLoaded_) {
+    LOG(ERROR) << "bpf programs are not loaded";
+    return kEmptyString.data();
+  }
+  auto sim = KatranSimulator(getKatranProgFd());
+  return sim.getRealForFlow(flow);
+}
+
 bool KatranLb::updateVipMap(
     const ModifyAction action,
     const VipKey& vip,

katran/lib/KatranLb.h

@@ -30,6 +30,7 @@
 #include "katran/lib/CHHelpers.h"
 #include "katran/lib/IpHelpers.h"
 #include "katran/lib/KatranLbStructs.h"
+#include "katran/lib/KatranSimulator.h"
 #include "katran/lib/Vip.h"
 
 namespace katran {
@@ -485,6 +486,16 @@ class KatranLb {
     return lbStats_;
   }
 
+  /**
+   * @param KatranFlow 5 tuple which describes a flow
+   * @return string address of the real.
+   *
+   * getRealForFlow functions returns address of the real where specified
+   * 5 tuple is going to be sent.
+   * returns empty string if given 5 tuple does not belong to a configured vip
+   */
+  const std::string getRealForFlow(const KatranFlow& flow);
+
  private:
   /**
    * update vipmap(add or remove vip) in forwarding plane

katran/lib/KatranSimulator.cpp

@@ -0,0 +1,241 @@
+#include "katran/lib/KatranSimulator.h"
+
+#include <folly/IPAddress.h>
+#include <glog/logging.h>
+#include <cstring>
+
+#include "katran/lib/BpfAdapter.h"
+
+extern "C" {
+#include <arpa/inet.h>
+#include <linux/if_ether.h>
+#include <linux/in.h>
+#include <linux/ip.h>
+#include <linux/ipv6.h>
+#include <linux/udp.h>
+}
+
+namespace katran {
+
+namespace {
+constexpr uint16_t kMaxXdpPcktSize = 4096;
+constexpr uint16_t kTestPacketSize = 512;
+constexpr int kTestRepeatCount = 1;
+constexpr uint8_t kDefaultTtl = 64;
+constexpr uint8_t kIPv6AddrSize = 16;
+constexpr folly::StringPiece kEmptyString = "";
+} // namespace
+
+namespace {
+
+void createV4Packet(
+    const folly::IPAddress& src,
+    const folly::IPAddress& dst,
+    std::unique_ptr<folly::IOBuf>& buf,
+    uint8_t proto,
+    uint16_t size) {
+  auto ehdr = reinterpret_cast<struct ethhdr*>(buf->writableData());
+  auto iph = reinterpret_cast<struct iphdr*>(
+      buf->writableData() + sizeof(struct ethhdr));
+  ehdr->h_proto = htons(ETH_P_IP);
+  iph->ihl = 5;
+  iph->version = 4;
+  iph->frag_off = 0;
+  iph->protocol = proto;
+  iph->check = 0;
+  iph->tos = 0;
+  iph->tot_len = htons(size);
+  iph->daddr = dst.asV4().toLong();
+  iph->saddr = src.asV4().toLong();
+  iph->ttl = kDefaultTtl;
+}
+
+void createV6Packet(
+    const folly::IPAddress& src,
+    const folly::IPAddress& dst,
+    std::unique_ptr<folly::IOBuf>& buf,
+    uint8_t proto,
+    uint16_t size) {
+  auto ehdr = reinterpret_cast<struct ethhdr*>(buf->writableData());
+  auto ip6h = reinterpret_cast<struct ipv6hdr*>(
+      buf->writableData() + sizeof(struct ethhdr));
+  ehdr->h_proto = htons(ETH_P_IPV6);
+  ip6h->version = 6;
+  ip6h->priority = 0;
+  std::memset(ip6h->flow_lbl, 0, sizeof(ip6h->flow_lbl));
+  ip6h->nexthdr = proto;
+  ip6h->payload_len = htons(size - sizeof(struct ipv6hdr));
+  ip6h->hop_limit = kDefaultTtl;
+  std::memcpy(
+      ip6h->daddr.s6_addr16, dst.asV6().toBinary().data(), kIPv6AddrSize);
+  std::memcpy(
+      ip6h->saddr.s6_addr16, src.asV6().toBinary().data(), kIPv6AddrSize);
+}
+
+void createTcpHeader(
+    std::unique_ptr<folly::IOBuf>& buf,
+    uint16_t srcPort,
+    uint16_t dstPort,
+    uint16_t offset) {
+  auto tcph = reinterpret_cast<struct tcphdr*>(buf->writableData() + offset);
+  std::memset(tcph, 0, sizeof(struct tcphdr));
+  tcph->source = htons(srcPort);
+  tcph->dest = htons(dstPort);
+  tcph->syn = 1;
+}
+
+void createUdpHeader(
+    std::unique_ptr<folly::IOBuf>& buf,
+    uint16_t srcPort,
+    uint16_t dstPort,
+    uint16_t offset,
+    uint16_t size) {
+  auto udph = reinterpret_cast<struct udphdr*>(buf->writableData() + offset);
+  std::memset(udph, 0, sizeof(struct udphdr));
+  udph->source = htons(srcPort);
+  udph->dest = htons(dstPort);
+  udph->len = size;
+}
+
+const std::string toV4String(uint32_t addr) {
+  return folly::IPAddressV4::fromLong(addr).str();
+}
+
+const std::string toV6String(uint8_t const* v6) {
+  folly::ByteRange bytes(v6, kIPv6AddrSize);
+  return folly::IPAddressV6::fromBinary(bytes).str();
+}
+
+std::string getPcktDst(std::unique_ptr<folly::IOBuf>& pckt) {
+  if (pckt->computeChainDataLength() < sizeof(struct ethhdr)) {
+    LOG(ERROR) << "resulting packet is invalid";
+    return kEmptyString.data();
+  }
+  const struct ethhdr* ehdr =
+      reinterpret_cast<const struct ethhdr*>(pckt->data());
+  if (ehdr->h_proto == htons(ETH_P_IP)) {
+    if (pckt->computeChainDataLength() <
+        (sizeof(struct ethhdr) + sizeof(struct iphdr))) {
+      LOG(ERROR) << "resulting ipv4 packet is invalid";
+      return kEmptyString.data();
+    }
+    const struct iphdr* iph = reinterpret_cast<const struct iphdr*>(
+        pckt->data() + sizeof(struct ethhdr));
+    return toV4String(iph->daddr);
+  } else {
+    if (pckt->computeChainDataLength() <
+        (sizeof(struct ethhdr) + sizeof(struct ipv6hdr))) {
+      LOG(ERROR) << "resulting ipv6 packet is invalid";
+      return kEmptyString.data();
+    }
+    const struct ipv6hdr* ip6h = reinterpret_cast<const struct ipv6hdr*>(
+        pckt->data() + sizeof(struct ethhdr));
+    return toV6String(ip6h->daddr.s6_addr);
+  }
+}
+
+std::unique_ptr<folly::IOBuf> createPacketFromFlow(const KatranFlow& flow) {
+  int offset = sizeof(struct ethhdr);
+  bool is_tcp = true;
+  bool is_v4 = true;
+  size_t l3hdr_len;
+
+  auto srcExp = folly::IPAddress::tryFromString(flow.src);
+  auto dstExp = folly::IPAddress::tryFromString(flow.dst);
+  if (srcExp.hasError() || dstExp.hasError()) {
+    LOG(ERROR) << "malformed src or dst ip address. src: " << flow.src
+               << " dst: " << flow.dst;
+    return nullptr;
+  }
+  auto src = srcExp.value();
+  auto dst = dstExp.value();
+  if (src.family() != dst.family()) {
+    LOG(ERROR) << "src and dst must have same address family";
+    return nullptr;
+  }
+  auto pckt = folly::IOBuf::create(kTestPacketSize);
+  if (!pckt) {
+    LOG(ERROR) << "cannot allocate IOBuf";
+    return pckt;
+  }
+  if (src.family() == AF_INET) {
+    l3hdr_len = sizeof(struct iphdr);
+  } else {
+    is_v4 = false;
+    l3hdr_len = sizeof(struct ipv6hdr);
+  }
+  offset += l3hdr_len;
+  switch (flow.proto) {
+    case IPPROTO_TCP:
+      break;
+    case IPPROTO_UDP:
+      is_tcp = false;
+      break;
+    default:
+      LOG(ERROR) << "unsupported protocol: " << flow.proto
+                 << " must be either TCP or UDP";
+      return nullptr;
+  }
+  pckt->append(kTestPacketSize);
+  auto payload_size = kTestPacketSize - sizeof(struct ethhdr);
+  if (is_v4) {
+    createV4Packet(src, dst, pckt, flow.proto, payload_size);
+  } else {
+    createV6Packet(src, dst, pckt, flow.proto, payload_size);
+  }
+  payload_size -= l3hdr_len;
+  if (is_tcp) {
+    createTcpHeader(pckt, flow.srcPort, flow.dstPort, offset);
+  } else {
+    createUdpHeader(pckt, flow.srcPort, flow.dstPort, offset, payload_size);
+  }
+  return pckt;
+}
+
+} // namespace
+
+KatranSimulator::KatranSimulator(int progFd) : progFd_(progFd) {}
+
+KatranSimulator::~KatranSimulator() {}
+
+std::unique_ptr<folly::IOBuf> KatranSimulator::runSimulation(
+    std::unique_ptr<folly::IOBuf> pckt) {
+  auto rpckt = folly::IOBuf::create(kMaxXdpPcktSize);
+  if (!rpckt) {
+    LOG(ERROR) << "not able to allocate memory for resulting packet";
+    return rpckt;
+  }
+  uint32_t output_pckt_size{0};
+  uint32_t prog_ret_val{0};
+  auto res = BpfAdapter::testXdpProg(
+      progFd_,
+      kTestRepeatCount,
+      pckt->writableData(),
+      pckt->computeChainDataLength(),
+      rpckt->writableData(),
+      &output_pckt_size,
+      &prog_ret_val);
+  if (res < 0) {
+    LOG(ERROR) << "failed to run simulator";
+    return nullptr;
+  }
+  if (prog_ret_val != XDP_TX) {
+    return nullptr;
+  }
+  rpckt->append(output_pckt_size);
+  return rpckt;
+}
+
+const std::string KatranSimulator::getRealForFlow(const KatranFlow& flow) {
+  auto pckt = createPacketFromFlow(flow);
+  if (!pckt) {
+    return kEmptyString.data();
+  }
+  auto rpckt = runSimulation(std::move(pckt));
+  if (!rpckt) {
+    return kEmptyString.data();
+  }
+  return getPcktDst(rpckt);
+}
+
+} // namespace katran

katran/lib/KatranSimulator.h

@@ -0,0 +1,54 @@
+#pragma once
+
+#include <folly/io/IOBuf.h>
+#include <memory>
+#include <string>
+
+namespace katran {
+
+/**
+ * struct that contains all the fields that uniquely identifies a flow
+ */
+struct KatranFlow {
+  // source ip address of the packet
+  std::string src;
+  // destination ip address of the packet
+  std::string dst;
+  uint16_t srcPort;
+  uint16_t dstPort;
+  // protocol number (e.g. 6 for TCP, 17 for UDP)
+  uint8_t proto;
+};
+
+/**
+ * KatranSimulator allows end user to simulate what is going to happen
+ * with specified packet after it is processed by katran load balancer.
+ * For e.g. where (address of the real) this packet is going to be sent
+ */
+class KatranSimulator final {
+ public:
+  /**
+   * @param int progFd descriptor of katran xdp program
+   */
+  explicit KatranSimulator(int progFd);
+  ~KatranSimulator();
+
+  /**
+   * @param KatranFlow& flow that we are interested in
+   * @return string ip address of the real (or empty string if packet will not
+   * be sent)
+   *
+   * getRealForFlow helps to determines where specific flow is going to be sent
+   * by returning ip address of the real
+   */
+  const std::string getRealForFlow(const KatranFlow& flow);
+
+ private:
+  // runSimulation takes packet (in iobuf represenation) and
+  // run it through katran bpf program. It returns a modified pckt, if the
+  // result was XDP_TX or nullptr otherwise.
+  std::unique_ptr<folly::IOBuf> runSimulation(
+      std::unique_ptr<folly::IOBuf> pckt);
+  int progFd_;
+};
+} // namespace katran