Adding load balance setups for both zmq and grpc

CMakeLists.txt

@@ -77,4 +77,8 @@ add_executable(try_grpc_server
 target_include_directories(try_grpc_server BEFORE PRIVATE lib)
 target_link_libraries(try_grpc_server ${CONAN_LIBS} flat_msg)
 
+add_executable(load_balance_benchmarks bench/load_balance_bench.cc)
+target_include_directories(load_balance_benchmarks BEFORE PRIVATE lib)
+target_link_libraries(load_balance_benchmarks ${CONAN_LIBS} flat_msg)
+
 set(CMAKE_RUNTIME_OUTPUT_DIRECTORY ${CMAKE_SOURCE_DIR}/bin/)

Makefile

@@ -1,11 +1,22 @@
-.DEFAULT_GOAL:= build-debug
+.DEFAULT_GOAL:= build-release
 
-clean:
+build-debug:
 	rm -rf cmake-build-debug
-
-
-build-debug: clean
 	mkdir -p cmake-build-debug/generated
 	cd cmake-build-debug && conan install .. && cd generated && flatc --cpp --scoped-enums --grpc ../../log_msg.fbs
 	cd cmake-build-debug && cmake -DCMAKE_BUILD_TYPE=Debug ..
 	cd cmake-build-debug && cmake --build . -- -j 4
+
+build-release:
+	rm -rf cmake-build-release
+	mkdir -p cmake-build-release/generated
+	cd cmake-build-release && conan install .. && cd generated && flatc --cpp --scoped-enums --grpc ../../log_msg.fbs
+	cd cmake-build-release && cmake -DCMAKE_BUILD_TYPE=Release ..
+	cd cmake-build-release && cmake --build . -- -j 4
+
+benchmarks: build-release
+	docker-compose -f ./load_balance/nginx_config/nginx-docker-compose.yml down
+	docker-compose -f ./load_balance/haconfig/ha-docker-compose.yml down
+	docker-compose -f ./load_balance/nginx_config/nginx-docker-compose.yml up -d
+	docker-compose -f ./load_balance/haconfig/ha-docker-compose.yml up -d
+	/bin/bash -c "cmake-build-release/bin/load_balance_benchmarks"

bench/benchmarks.cc

@@ -73,25 +73,6 @@ static void runzmq(benchmark::State &state)
 
 BENCHMARK(runzmq)->Iterations(150000);
 
-// You need to have grpc_server running
-static void rungrpc(benchmark::State &state)
-{
-  std::string server_address("localhost:50051");
-
-  auto channel = grpc::CreateChannel(server_address,
-                                     grpc::InsecureChannelCredentials());
-  LogClientImpl greeter(channel);
-
-  for (auto _ : state)
-  {
-    greeter.send_log(Logger::LogLevel::INSANE, __CODE_POINT__,
-                     "hello from msg2", "this service", "hostname");
-    greeter.send_log(Logger::LogLevel::INSANE, __CODE_POINT__,
-                     "hello from msg2", "this service", "hostname");
-  }
-}
-//BENCHMARK(rungrpc);
-
 // You need to have grpc_server running
 static void rungrpc2(benchmark::State &state)
 {
@@ -116,8 +97,6 @@ static void rungrpc2(benchmark::State &state)
 BENCHMARK(rungrpc2)->Iterations(150000);
 
 // Run the benchmark
-//BENCHMARK_MAIN();
-
 int main(int argc, char **argv)
 {
   FLAGS_log_dir = ".";

bench/load_balance_bench.cc

@@ -0,0 +1,146 @@
+//
+// Created by zerogap on 12/12/21.
+//
+#include <benchmark/benchmark.h>
+
+#include <thread>
+#include <zmq_addon.hpp>
+
+#include "LogEntrySerializer.h"
+#include "log_client.h"
+
+#define STR_HELPER(x) #x
+#define STR(x) STR_HELPER(x)
+#define __CODE_POINT__ (__FILE__ "::" STR(__LINE__))
+
+static void log_creation(benchmark::State &state)
+{
+  for (auto _ : state)
+  {
+    LogEntrySerializer new_log(Logger::LogLevel::DEBUG, __CODE_POINT__, "hello",
+                               "this servec", "hostname");
+    new_log.export_fb();
+    LogEntrySerializer new_log2(Logger::LogLevel::INSANE, __CODE_POINT__,
+                                "hello from msg2", "this service", "hostname");
+    new_log2.export_fb();
+  }
+}
+BENCHMARK(log_creation)->Iterations(150000);
+
+zmq::context_t ctx(1);
+zmq::socket_t sock1(ctx, zmq::socket_type::push);
+zmq::socket_t sock2(ctx, zmq::socket_type::push);
+zmq::socket_t sock3(ctx, zmq::socket_type::push);
+uint64_t msg_count{ 0 };
+static void runzmq(benchmark::State &state)
+{
+  for (auto _ : state)
+  {
+    ++msg_count;
+    LogEntrySerializer new_log(Logger::LogLevel::DEBUG, __CODE_POINT__,
+                               "hello from msg " + std::to_string(msg_count),
+                               "this servec", "hostname");
+    new_log.export_fb();
+
+    ++msg_count;
+    LogEntrySerializer new_log2(Logger::LogLevel::INSANE, __CODE_POINT__,
+                                "hello from msg " + std::to_string(msg_count),
+                                "this "
+                                "service",
+                                "hostname");
+    new_log2.export_fb();
+
+    ++msg_count;
+    LogEntrySerializer new_log3(Logger::LogLevel::INSANE, __CODE_POINT__,
+                                "hello from msg " + std::to_string(msg_count),
+                                "this "
+                                "service",
+                                "hostname");
+    new_log3.export_fb();
+
+    sock1.send(zmq::const_buffer(zmq::buffer(new_log.get_data())),
+               zmq::send_flags::dontwait);
+    sock2.send(zmq::const_buffer(zmq::buffer(new_log2.get_data())),
+               zmq::send_flags::dontwait);
+    sock3.send(zmq::const_buffer(zmq::buffer(new_log3.get_data())),
+               zmq::send_flags::dontwait);
+  }
+}
+
+BENCHMARK(runzmq)->Iterations(150000);
+
+// You need to have nginx load balance compose
+// Messages will be load balanced between all grpc servers
+uint64_t grpc_lb_sent{0};
+static void rungrpc(benchmark::State &state)
+{
+  std::string server_address("localhost:50052");
+
+  auto channel = grpc::CreateChannel(server_address,
+                                     grpc::InsecureChannelCredentials());
+  LogClientImpl greeter(channel);
+
+  for (auto _ : state)
+  {
+    ++grpc_lb_sent;
+    greeter.send_log(Logger::LogLevel::INSANE, __CODE_POINT__,
+                     "hello from msg "+ std::to_string(grpc_lb_sent), "this service",
+                     "hostname");
+    ++grpc_lb_sent;
+    greeter.send_log(Logger::LogLevel::INSANE, __CODE_POINT__,
+                     "hello from msg "+ std::to_string(grpc_lb_sent), "this service", "hostname");
+    ++grpc_lb_sent;
+    greeter.send_log(Logger::LogLevel::INSANE, __CODE_POINT__,
+                     "hello from msg "+ std::to_string(grpc_lb_sent), "this service", "hostname");
+  }
+}
+BENCHMARK(rungrpc)->Iterations(150000);
+
+// Run the benchmark
+
+void setup_socks(){
+
+  // Need to have the ha load balancer running first
+  sock1.connect(
+          "tcp://127.0.0.1:12404");
+  sock1.set(zmq::sockopt::immediate, 1);
+  sock1.set(zmq::sockopt::sndhwm, 1024);
+  sock1.set(zmq::sockopt::reconnect_ivl_max, 2500);
+  sock1.set(zmq::sockopt::linger, 500);
+
+  sock2.connect(
+          "tcp://127.0.0.1:12404");
+  sock2.set(zmq::sockopt::immediate, 1);
+  sock2.set(zmq::sockopt::sndhwm, 1024);
+  sock2.set(zmq::sockopt::reconnect_ivl_max, 2500);
+  sock2.set(zmq::sockopt::linger, 500);
+
+  sock3.connect(
+          "tcp://127.0.0.1:12404");
+  sock3.set(zmq::sockopt::immediate, 1);
+  sock3.set(zmq::sockopt::sndhwm, 1024);
+  sock3.set(zmq::sockopt::reconnect_ivl_max, 2500);
+  sock3.set(zmq::sockopt::linger, 500);
+}
+
+int main(int argc, char **argv)
+{
+  FLAGS_log_dir = ".";
+  FLAGS_logbufsecs = 0;
+  google::InitGoogleLogging("benchmark.log");
+  setup_socks();
+  ::benchmark::Initialize(&argc, argv);
+  if (::benchmark::ReportUnrecognizedArguments(argc, argv)) return 1;
+  ::benchmark::RunSpecifiedBenchmarks();
+  std::cout << "msgs sent from zmq: " << msg_count << std::endl;
+  std::cout << "msgs sent to load balancer: " << grpc_lb_sent << std::endl;
+  ctx.shutdown();
+  std::cout << "zmq ctx shutdown\n";
+  sock1.close();
+  std::cout << "Socket disconnected\n";
+  google::ShutdownGoogleLogging();
+  std::cout << "Google logging shut down\n";
+  ::benchmark::Shutdown();
+  std::cout << "benchmark shutdown\n";
+  return 0;
+}

load_balance/haconfig/conf/haproxy.cfg

@@ -0,0 +1,63 @@
+#---------------------------------------------------------------------
+# Global settings
+#---------------------------------------------------------------------
+global
+    log         127.0.0.1 local2
+
+    chroot      /var/lib/haproxy
+    pidfile     /var/run/haproxy.pid
+    maxconn     4000
+    user        haproxy
+    group       haproxy
+    daemon
+
+    # turn on stats unix socket
+    stats socket /var/lib/haproxy/stats
+
+#---------------------------------------------------------------------
+# common defaults that all the 'listen' and 'backend' sections will
+# use if not designated in their block
+#---------------------------------------------------------------------
+defaults
+    mode                    tcp
+    log                     global
+    option                  httplog
+    option                  dontlognull
+    option http-server-close
+    #option forwardfor       except 127.0.0.0/8
+    option                  redispatch
+    retries                 3
+    timeout http-request    10s
+    timeout queue           1m
+    timeout connect         10s
+    timeout client          1m
+    timeout server          1m
+    timeout http-keep-alive 10s
+    timeout check           10s
+    maxconn                 300000
+
+frontend stats
+  bind *:8404
+  stats enable
+  stats uri /
+  stats refresh 10s
+
+frontend rserve_frontend
+    bind *:81
+    mode tcp
+    option tcplog
+    timeout client  1m
+    default_backend rserve_backend
+
+backend rserve_backend
+    mode tcp
+#     option tcplog # ignored
+    option log-health-checks
+    option redispatch
+    log global
+    balance roundrobin
+    timeout connect 10s
+    timeout server 1m
+    server rserve1 zmq_consumer_1:52236 check
+    server rserve2 zmq_consumer_2:52236 check
+    server rserve3 zmq_consumer_3:52236 check

load_balance/haconfig/ha-docker-compose.yml

@@ -0,0 +1,50 @@
+version: '3'
+services:
+  ha-proxy:
+    image: haproxytech/haproxy-alpine:2.4
+    container_name: haproxy
+    ports:
+      - "12404:81"
+      - "8404:8404"
+    volumes:
+      - ./conf:/usr/local/etc/haproxy:ro
+    networks:
+      - zmq_ingressor
+    depends_on:
+      - zmq_consumer_1
+      - zmq_consumer_2
+      - zmq_consumer_3
+
+  zmq_consumer_1:
+    image: ubuntu:20.04
+    container_name: zmq_consumer_1
+    volumes:
+      - ../../cmake-build-release/bin:/home/zero/bin
+    command:
+      ["/bin/bash", "-c", "/home/zero/bin/zmq_server"]
+    networks:
+      - zmq_ingressor
+
+  zmq_consumer_2:
+    image: ubuntu:20.04
+    container_name: zmq_consumer_2
+    volumes:
+      - ../../cmake-build-release/bin:/home/zero/bin
+    command:
+      ["/bin/bash", "-c", "/home/zero/bin/zmq_server"]
+    networks:
+      - zmq_ingressor
+
+  zmq_consumer_3:
+    image: ubuntu:20.04
+    container_name: zmq_consumer_3
+    volumes:
+      - ../../cmake-build-release/bin:/home/zero/bin
+    command:
+      ["/bin/bash", "-c", "/home/zero/bin/zmq_server"]
+    networks:
+      - zmq_ingressor
+
+
+networks:
+  zmq_ingressor:

load_balance/nginx_config/nginx-docker-compose.yml

@@ -0,0 +1,55 @@
+version: '3'
+services:
+  nginx:
+#    image: tekn0ir/nginx-stream
+#    image: hokaccha/grpc-web-proxy-nginx:latest
+    image: nginx:1.20.0
+    container_name: production_nginx
+    volumes:
+      - ./nginx.conf:/etc/nginx/nginx.conf
+    ports:
+      - "52200:52200"
+      - "50052:50052"
+      - "8080:8080"
+    depends_on:
+      - grpc
+      - grpc1
+      - grpc2
+#    command: /bin/sh -c "envsubst '$$NGINX_GRPC_PASS' < /etc/nginx/conf.d/grpc_proxy.conf.template > /etc/nginx/conf.d/grpc_proxy.conf && nginx -g 'daemon off;'"
+#    environment:
+#      NGINX_GRPC_PASS: 'grpc_consumer_1:8080'
+    networks:
+      - zmq_ingressor
+
+  grpc:
+    image: ubuntu:20.04
+    container_name: grpc
+    volumes:
+      - ../../cmake-build-release/bin:/home/zero/bin
+    command:
+      ["/bin/bash", "-c", "/home/zero/bin/try_grpc_server"]
+    networks:
+      - zmq_ingressor
+
+  grpc1:
+    image: ubuntu:20.04
+    container_name: grpc1
+    volumes:
+      - ../../cmake-build-release/bin:/home/zero/bin
+    command:
+      ["/bin/bash", "-c", "/home/zero/bin/try_grpc_server"]
+    networks:
+      - zmq_ingressor
+
+  grpc2:
+    image: ubuntu:20.04
+    container_name: grpc2
+    volumes:
+      - ../../cmake-build-release/bin:/home/zero/bin
+    command:
+      ["/bin/bash", "-c", "/home/zero/bin/try_grpc_server"]
+    networks:
+      - zmq_ingressor
+
+networks:
+  zmq_ingressor: