OmniWindow for AE

Signed-off-by: kira <sunhaifeng@stu.pku.edu.cn>

.gitignore

@@ -0,0 +1,2 @@
+.DS_Store
+build/*

CMakeLists.txt

@@ -0,0 +1,73 @@
+cmake_minimum_required(VERSION 3.6)
+
+option(IS_SWITCH "Is building switch programs" OFF)
+option(ENABLE_EXP_DC "Enable building the micro-benchmarks of data collection" OFF)
+
+if(${IS_SWITCH})
+  project(OmniWindow LANGUAGES)
+  message(STATUS "Building switch programs.")
+else()
+  project(OmniWindow)
+  message(STATUS "Building host programs.")
+endif()
+
+if(NOT ${IS_SWITCH})
+  set(CMAKE_CXX_STANDARD 17)
+
+  add_compile_options(-mssse3 -fcommon -march=native)
+
+  include_directories(include)
+  add_subdirectory(lib)
+  add_subdirectory(tools)
+endif()
+
+function(add_p4 name source compile_flags py_script)
+  if(${IS_SWITCH})
+    make_directory("${CMAKE_CURRENT_BINARY_DIR}/${name}.build")
+
+    add_custom_command(
+      OUTPUT "${CMAKE_CURRENT_BINARY_DIR}/${name}.sh"
+      # COMMAND rm -rf *
+      COMMAND cmake \"$ENV{SDE}/p4studio\"
+        -DCMAKE_INSTALL_PREFIX=\"$ENV{SDE_INSTALL}\"
+        -DCMAKE_MODULE_PATH=\"$ENV{SDE}/cmake\"
+        -DP4_NAME=\"omniwin-${name}\"
+        -DP4_PATH=\"${CMAKE_CURRENT_SOURCE_DIR}/${source}\"
+        ${compile_flags}
+      COMMAND make
+      COMMAND make install
+      COMMAND echo \"\#\! /bin/bash -i\" >"${CMAKE_CURRENT_BINARY_DIR}/${name}.sh"
+      COMMAND echo \"
+          if \[\[ ! -a "${CMAKE_SOURCE_DIR}/switch-shell.txt" \]\]\; then
+            echo "Missing file ${CMAKE_SOURCE_DIR}/switch-shell.txt" \;
+            exit\;
+          fi\;
+          cd \\$$SDE\;
+          ./run_switchd.sh -p "omniwin-${name}" -- --background &
+          ./run_bfshell.sh -f "${CMAKE_SOURCE_DIR}/switch-shell.txt" \;
+          if \[\[ -a "${CMAKE_SOURCE_DIR}/switch-config.json" \]\]\; then
+            cp "${CMAKE_SOURCE_DIR}/switch-config.json" /tmp/OmniWindow-config.json\;
+            ./run_bfshell.sh -i -b "${CMAKE_CURRENT_SOURCE_DIR}/${py_script}" \;
+          else
+            echo "Missing file ${CMAKE_SOURCE_DIR}/switch-config.json" \;
+            echo "Press Ctrl-C to quit..." \;
+          fi\;
+          fg
+        \" >>"${CMAKE_CURRENT_BINARY_DIR}/${name}.sh"
+      COMMAND chmod +x "${CMAKE_CURRENT_BINARY_DIR}/${name}.sh"
+      WORKING_DIRECTORY "${CMAKE_CURRENT_BINARY_DIR}/${name}.build"
+      DEPENDS "${CMAKE_CURRENT_SOURCE_DIR}/${source}"
+      DEPENDS "${CMAKE_CURRENT_SOURCE_DIR}/${py_script}"
+    )
+
+    add_custom_target(${name} ALL
+      DEPENDS "${CMAKE_CURRENT_BINARY_DIR}/${name}.sh"
+    )
+  endif()
+endfunction()
+
+add_subdirectory(apps)
+
+if(${ENABLE_EXP_DC})
+  add_subdirectory(Exp-dc)
+endif()

Exp-1/.gitignore

@@ -0,0 +1,4 @@
+.vscode/
+build/
+
+data

Exp-1/CMakeLists.txt

@@ -0,0 +1,13 @@
+project(OmniWindow-sim)
+
+cmake_minimum_required(VERSION 3.5)
+
+add_compile_options(-march=native)
+
+set(CMAKE_RUNTIME_OUTPUT_DIRECTORY bin)
+set(CMAKE_CXX_STANDARD 17)
+
+include_directories(include)
+
+add_subdirectory(lib)
+add_subdirectory(apps)

Exp-1/README.md

@@ -0,0 +1,36 @@
+# (Exp #1) Query-driven telemetry
+
+You can run the simulation on any host.
+
+## Requirement
+
+- `libcrcutil`:
+  ```sh
+  sudo apt install libcrcutil-dev
+  ```
+
+## Build
+
+```sh
+# Make sure you are at the ${OmniWindow_DIR}/Exp-1/ directory.
+mkdir build && cd build
+cmake .. -DCMAKE_BUILD_TYPE=Release
+make -j
+cd ..
+```
+
+## Experiement
+
+1. Run the experiment:
+   ```sh
+   # Make sure you are at the ${OmniWindow_DIR}/Exp-1/ directory.
+   . run.sh
+   ```
+   and wait for all jobs to finish (you can see their status with command `jobs`, or wait for all jobs with command `wait`).
+
+2. Merge the results in a table:
+   ```sh
+   # Make sure you are at the ${OmniWindow_DIR}/Exp-1/ directory.
+   python3 merge-result.py result
+   ```
+   Then the results are in `${OmniWindow_DIR}/Exp-1/result/recall.csv` and `${OmniWindow_DIR}/Exp-1/result/precision.csv`.

Exp-1/apps/CMakeLists.txt

@@ -0,0 +1,12 @@
+find_package(Threads REQUIRED)
+link_libraries(Threads::Threads)
+link_libraries(omniwindow-sim)
+
+add_executable(new_tcp_connections new_tcp_connections.cpp)
+add_executable(ssh_brute ssh_brute.cpp)
+add_executable(superspreader superspreader.cpp)
+add_executable(port_scan port_scan.cpp)
+add_executable(ddos ddos.cpp)
+add_executable(syn_flood syn_flood.cpp)
+add_executable(completed_flow completed_flow.cpp)
+add_executable(slowloris_attack slowloris_attack.cpp)

Exp-1/apps/completed_flow.cpp

@@ -0,0 +1,69 @@
+#include "Query.hpp"
+#include "Reduce.hpp"
+#include "utils.hpp"
+
+#include "evaluate.hpp"
+
+using Key = uint32_t;
+
+class CompletedFlow : public Query<Key> {
+  int32_t threshold;
+  std::unique_ptr<Reduce<Key, uint32_t>> reduce1, reduce2;
+
+public:
+  std::vector<int32_t> vals;
+
+  CompletedFlow(uint32_t threshold_,
+        std::unique_ptr<Reduce<Key, uint32_t>> reduce1_,
+        std::unique_ptr<Reduce<Key, uint32_t>> reduce2_) :
+      threshold(threshold_), reduce1(std::move(reduce1_)),
+      reduce2(std::move(reduce2_)) {}
+
+  std::optional<HashSet<Key>> switchWin() override {
+    auto r1 = reduce1->switchWin();
+    auto r2 = reduce2->switchWin();
+    if (r1 && r2) {
+      HashSet<Key> res;
+      for (auto [key, v1] : *r1) {
+        int32_t v2 = r2->count(key) ? (*r2)[key] : 0;
+        int32_t v = (int32_t)v1 - v2;
+        vals.push_back(v);
+        if (v >= threshold)
+          res.insert(key);
+      }
+      return res;
+    }
+    return std::nullopt;
+  }
+
+  void process(PktInfo pkt) override {
+    if (pkt.key.proto == 6) {
+      if (pkt.tcp_flag == 2)
+        reduce1->process(pkt.key.dst_ip, 1);
+      if (pkt.tcp_flag & 1)
+        reduce2->process(pkt.key.dst_ip, 1);
+    }
+  }
+};
+
+int main(int argc, char **argv) {
+  int32_t threshold;
+
+  if (argc != 3 || sscanf(argv[2], "%d", &threshold) != 1) {
+    fprintf(stderr, "%s <trace> <threshold>\n", argv[0]);
+    return 1;
+  }
+
+  auto qryGen = [threshold]([[maybe_unused]] auto disGen, auto redGen) {
+    return std::make_unique<CompletedFlow>(
+      threshold,
+      redGen(),
+      redGen());
+  };
+
+  printf("Threshold: %d\n\n", threshold);
+
+  evaluate<Key, VoidDKey, Key, uint32_t>(argv[1], qryGen);
+
+  return 0;
+}

Exp-1/apps/ddos.cpp

@@ -0,0 +1,78 @@
+#include "Query.hpp"
+#include "Distinct.hpp"
+#include "Reduce.hpp"
+#include "utils.hpp"
+
+#include "evaluate.hpp"
+
+using Key = uint32_t;
+
+struct DKey {
+  uint32_t dst_ip;
+  uint32_t src_ip;
+} __attribute__((packed));
+
+class DDoS : public Query<Key> {
+  uint32_t threshold;
+  std::unique_ptr<Distinct<DKey>> distinct;
+  std::unique_ptr<Reduce<Key, uint32_t>> reduce;
+
+public:
+  std::vector<uint32_t> vals;
+
+  DDoS(uint32_t threshold_, std::unique_ptr<Distinct<DKey>> distinct_,
+        std::unique_ptr<Reduce<Key, uint32_t>> reduce_) :
+      threshold(threshold_), distinct(std::move(distinct_)), reduce(std::move(reduce_)) {}
+
+  std::optional<HashSet<Key>> switchWin() override {
+    distinct->switchWin();
+    if (auto r = reduce->switchWin()) {
+      HashSet<Key> res;
+      for (auto [key, val] : *r) {
+        vals.push_back(val);
+        if (val >= threshold)
+          res.insert(key);
+      }
+      return res;
+    }
+    return std::nullopt;
+  }
+
+  void process(PktInfo pkt) override {
+    DKey dkey{pkt.key.dst_ip, pkt.key.src_ip};
+    Key key = pkt.key.dst_ip;
+
+    if (auto *sDistinct = dynamic_cast<SlidingBaselineDistinct<DKey> *>(distinct.get()))
+      if (auto *sReduce = dynamic_cast<SlidingReduce<Key, uint32_t> *>(reduce.get())) {
+        size_t prevTtl = sDistinct->getTtlAndProcess(dkey);
+        sReduce->processWithPrevTtl(key, 1, prevTtl);
+        return;
+      }
+
+    if (distinct->process(dkey))
+      reduce->process(key, 1);
+  }
+};
+
+int main(int argc, char **argv) {
+  uint32_t threshold;
+
+  if (argc != 3 || sscanf(argv[2], "%u", &threshold) != 1) {
+    fprintf(stderr, "%s <trace> <threshold>\n", argv[0]);
+    return 1;
+  }
+
+  auto qryGen = [threshold](auto disGen, auto redGen) {
+    return std::make_unique<DDoS>(
+      threshold,
+      disGen(),
+      redGen()
+    );
+  };
+
+  printf("Threshold: %u\n\n", threshold);
+
+  evaluate<Key, DKey, Key, uint32_t>(argv[1], qryGen);
+
+  return 0;
+}

Exp-1/apps/evaluate.hpp

@@ -0,0 +1,82 @@
+#ifndef EVALUATE_HPP
+#define EVALUATE_HPP
+
+#include <functional>
+#include <memory>
+#include <future>
+#include <type_traits>
+
+#include "Trace.h"
+#include "Query.hpp"
+
+#include "Distinct.hpp"
+#include "Reduce.hpp"
+
+struct VoidDKey {};
+
+template <typename Key, typename DKey, typename RKey, typename RVal, typename TQryGen>
+void evaluate(const char *tracePath, TQryGen qryGen, size_t distinctLen = 65536 * 32, size_t reduceLen = 65536) {
+
+  size_t nEpoch = 100; double interval = 0.5; double dcTime = 20e-3;
+  size_t nSlide = 5;
+
+  double slide = interval / (double)nSlide;
+
+  Trace trace = readTrace(tracePath);
+
+  auto fResISW = std::async(std::launch::async, getResult<Key>,
+      trace, nEpoch * nSlide, slide, 0,
+      qryGen(
+        [nSlide]() {
+          return std::make_unique<SlidingBaselineDistinct<DKey>>(nSlide); },
+        [nSlide]() {
+          return std::make_unique<SlidingReduce<RKey, RVal>>(
+            nSlide, std::make_unique<BaselineReduce<RKey, RVal>>()); }));
+
+  auto fResTW1 = std::async(std::launch::async, getResult<Key>,
+      trace, nEpoch, interval, dcTime,
+      qryGen(
+        [distinctLen] { return std::make_unique<DataPlaneDistinct<DKey>>(distinctLen * 4); },
+        [reduceLen] { return std::make_unique<DataPlaneReduce<RKey, RVal>>(reduceLen * 4); }));
+
+  auto fResTW2 = std::async(std::launch::async, getResult<Key>,
+      trace, nEpoch, interval, 0,
+      qryGen(
+        [distinctLen]() { return std::make_unique<DataPlaneDistinct<DKey>>(distinctLen * 4); },
+        [reduceLen]() { return std::make_unique<DataPlaneReduce<RKey, RVal>>(reduceLen * 4); }));
+
+  auto fResOTW = std::async(std::launch::async, getResult<Key>,
+      trace, nEpoch * nSlide, slide, 0,
+      qryGen(
+        [nSlide, distinctLen]() {
+          return std::make_unique<FineGrainedLargeDistinct<DKey>>(
+            nSlide, std::make_unique<DataPlaneDistinct<DKey>>(distinctLen)); },
+        [nSlide, reduceLen]() {
+          return std::make_unique<FineGrainedReduce<RKey, RVal>>(
+            nSlide, std::make_unique<DataPlaneReduce<RKey, RVal>>(reduceLen)); }));
+
+  auto fResITW = std::async(std::launch::async, getResult<Key>,
+    trace, nEpoch, interval, 0,
+    qryGen(
+      []() { return std::make_unique<BaselineDistinct<DKey>>(); },
+      []() { return std::make_unique<BaselineReduce<RKey, RVal>>(); }));
+
+  auto fResOSW = std::async(std::launch::async, getResult<Key>,
+      trace, nEpoch * nSlide, slide, 0,
+      qryGen(
+        [nSlide, distinctLen]() {
+          return std::make_unique<FineGrainedLargeDistinct<DKey>>(
+            nSlide, std::make_unique<DataPlaneDistinct<DKey>>(distinctLen)); },
+        [nSlide, reduceLen]() {
+          return std::make_unique<SlidingReduce<RKey, RVal>>(
+            nSlide, std::make_unique<DataPlaneReduce<RKey, RVal>>(reduceLen)); }));
+
+  auto mergedISW = collapseResultAll(fResISW.get());
+  ResultInfo<Key>(mergedISW, collapseResultAll(fResTW1.get())).print("TW1");
+  ResultInfo<Key>(mergedISW, collapseResultAll(fResTW2.get())).print("TW2");
+  ResultInfo<Key>(mergedISW, collapseResultAll(fResOTW.get())).print("OTW");
+  ResultInfo<Key>(mergedISW, collapseResultAll(fResITW.get())).print("ITW");
+  ResultInfo<Key>(mergedISW, collapseResultAll(fResOSW.get())).print("OSW");
+}
+
+#endif