Adding network latency simulator for block live stream consumers (#34)

Signed-off-by: Alfredo Gutierrez <[email protected]>

README.md

@@ -33,8 +33,16 @@ Please do not file a public ticket mentioning the vulnerability. Refer to the se
 # Running Locally
 
 1) Create a local temp directory.  For example, use `mktemp -d -t block-stream-temp-dir` to create a directory
-2) export BLOCKNODE_STORAGE_ROOT_PATH=<path to the temp directory> # You can add this to your .zshrc, etc
+2) Configuration variables
+```
+export BLOCKNODE_STORAGE_ROOT_PATH=<path to the temp directory> # You can add this to your .zshrc, etc
+```
+3) Optional Configuration variables
+```
+export BLOCKNODE_SERVER_CONSUMER_TIMEOUT_THRESHOLD="<NumberInMiliseconds>" #Default is 1500
+```
+
 3) ./gradlew run  # ./gradlew run --debug-jvm to run in debug mode
 
 # Running Tests
-1) ./gradlew build
+1) ./gradlew build

server/src/main/java/com/hedera/block/server/Constants.java

@@ -24,6 +24,7 @@ private Constants() {}
 
     // Config Constants
     public static final String BLOCKNODE_STORAGE_ROOT_PATH_KEY = "blocknode.storage.root.path";
+    public static final String BLOCKNODE_SERVER_CONSUMER_TIMEOUT_THRESHOLD_KEY = "blocknode.server.consumer.timeout.threshold";
 
     // Constants specified in the service definition of the .proto file
     public static final String SERVICE_NAME = "BlockStreamGrpc";

server/src/main/java/com/hedera/block/server/Server.java

@@ -59,11 +59,12 @@ public static void main(final String[] args) {
             final Config config = Config.create();
             Config.global(config);
 
+            // Get Timeout threshold from configuration
+            final long consumerTimeoutThreshold = config.get(BLOCKNODE_SERVER_CONSUMER_TIMEOUT_THRESHOLD_KEY).asLong().orElse(1500L);
+
             // Initialize the block storage, cache, and service
             final BlockStorage<BlockStreamServiceGrpcProto.Block> blockStorage = new FileSystemBlockStorage(BLOCKNODE_STORAGE_ROOT_PATH_KEY, config);
-
-            // TODO: Make timeoutThresholdMillis configurable
-            final BlockStreamService blockStreamService = new BlockStreamService(1500,
+            final BlockStreamService blockStreamService = new BlockStreamService(consumerTimeoutThreshold,
                     new LiveStreamMediatorImpl(new WriteThroughCacheHandler(blockStorage)));
 
             // Start the web server

server/src/test/network-latency-simulator/.gitignore

@@ -0,0 +1 @@
+/test-context

server/src/test/network-latency-simulator/Dockerfile

@@ -0,0 +1,27 @@
+# Use an official Ubuntu base image
+FROM ubuntu:latest
+
+# Set environment variables
+ENV DEBIAN_FRONTEND=noninteractive
+ENV GRPC_SERVER="host.docker.internal:8080"
+ENV GRPC_METHOD="BlockStreamGrpc/StreamSource"
+ENV PATH_TO_PROTO="/usr/local/protos/blockstream.proto"
+ENV PROTO_IMPORT_PATH="/usr/local/protos"
+ENV INITIAL_LATENCY=500
+ENV JITTER=500
+ENV BANDWIDTH=64
+ENV INCREASE_TIME=10
+ENV MAX_LATENCY=12000
+
+# Install required packages
+RUN apt-get update && \
+    apt-get install -y iproute2 iputils-ping curl net-tools iperf3 iptables kmod && \
+    curl -L https://github.com/fullstorydev/grpcurl/releases/download/v1.8.7/grpcurl_1.8.7_linux_x86_64.tar.gz -o grpcurl.tar.gz && \
+    tar -xvf grpcurl.tar.gz && mv grpcurl /usr/local/bin/grpcurl && rm grpcurl.tar.gz
+
+# Copy scripts and protos folder into the container
+COPY configure_latency.sh start.sh test-context/consumer.sh /usr/local/bin/
+COPY test-context/protos /usr/local/protos
+
+# Default command to run when starting the container
+CMD ["bash", "-c", "start.sh"]

server/src/test/network-latency-simulator/README.md

@@ -0,0 +1,59 @@
+# Network Latency Simulator
+
+Due to the asynchronous nature of the Streaming Service, it is important to test the system under different network conditions, such as high latency, packet loss, and jitter. This tool allows you to simulate different network conditions by adding latency, packet loss, low bandwidth and jitter to the network traffic.
+
+And making sure that a single `consumer` that is experiencing network issues does not affect the other `consumers` that are consuming the same stream from the BlockNode.
+
+This test aims to make sure that the system is resilient to network issues and that a single consumer that is experiencing network issues does not affect the other consumers that are consuming the same stream from the BlockNode.
+
+## Running Locally
+
+1. Move to the `network-latency-simulator` directory.
+2. Prepare the Test Context, Build the Docker Image with the network latency simulator.
+```bash
+cd server/src/test/network-latency-simulator
+
+./setup.sh
+
+docker build -t network-latency-simulator .
+```
+
+3. Start the BlockNode Server. (Follow instructions on main README.md)
+   - Due to the Latency, the consumers might be disconnected from the BlockNode, since the current timeout is 1500 ms, you should increase the timeout to 100000ms to be able to correctly test the network issues. (see main README.md of the server for more details on how to change the timeout)
+4. Start the producer and a single consumer (this consumer will be the control one without any network issues).
+```bash
+/server/src/test/resources/producer.sh 1 1000 # this will produce 1000 blocks
+/server/src/test/resources/consumer.sh 1 1000 # this will consume 1000 blocks
+```
+5. Start the consumer inside the network latency simulator container, you can start as many as you want.
+```bash
+docker run -it --cap-add=NET_ADMIN network-latency-simulator
+```
+
+The consumer inside the container will start consuming the blocks from the BlockNode, and you can see the network issues being simulated.
+The network latency simulator will simulate the following network issues:
+- Latency, increases every 10 seconds (by default) by 1000ms
+- Packet Loss (Drops 10% of the packets)
+- Low Bandwidth, limits the bandwidth to 64kbps.
+- Jitter, adds 500ms of jitter (latency variability) to the network.
+
+There are some environment variables that you can set to change the behavior of the network latency simulator:
+
+**configure_latency.sh:**
+- `LATENCY_INCREASE_INTERVAL`: The interval in seconds to increase the latency, default is 10 seconds.
+- `INITIAL_LATENCY`: The initial latency to start with, default is 500ms, once the MAX latency is reached, it will reset to the initial latency.
+- `JITTER`: The jitter to add to the network, default is 500ms.
+- `BANDWIDTH`: The bandwidth to limit the network to, default is 64kbps.
+- `INCREASE_TIME`: The time in seconds to increase the latency, default is 10 (seconds).
+- `MAX_LATENCY`: The maximum latency to reach, default is 12000 (ms).
+
+**consumer.sh:**
+- `GRPC_SERVER`: The gRPC server to connect to, default is `host.docker.internal:8080`, connects to the host BlockNode.
+- `GRPC_METHOD`: The gRPC method to call, default is `BlockStreamGrpc/StreamSource`.
+- `PATH_TO_PROTO`: The path to the proto file, default is `/usr/local/protos/blockstream.proto` (inside the container).
+- `PROTO_IMPORT_PATH`: The import path of the proto file, default is `/usr/local/protos` (inside the container).
+
+Example of how to set the environment variables when running the container:
+```bash
+docker run -it --cap-add=NET_ADMIN -e LATENCY_INCREASE_INTERVAL=5 -e INITIAL_LATENCY=1000 -e JITTER=1000 -e BANDWIDTH=128 -e INCREASE_TIME=5 -e MAX_LATENCY=10000 network-latency-simulator
+```

server/src/test/network-latency-simulator/configure_latency.sh

@@ -0,0 +1,42 @@
+#!/bin/bash
+
+# Default values
+DEFAULT_INITIAL_LATENCY=1000
+DEFAULT_JITTER=500
+DEFAULT_BANDWIDTH=64
+DEFAULT_INCREASE_TIME=10
+MAX_LATENCY=12000
+PACKET_LOSS=5  # Packet loss in percentage
+
+# Parameters with default values
+INITIAL_LATENCY=${INITIAL_LATENCY:-$DEFAULT_INITIAL_LATENCY}
+JITTER=${JITTER:-$DEFAULT_JITTER}
+BANDWIDTH=${BANDWIDTH:-$DEFAULT_BANDWIDTH}
+INCREASE_TIME=${INCREASE_TIME:-$DEFAULT_INCREASE_TIME}
+CURRENT_LATENCY=$INITIAL_LATENCY
+MAX_LATENCY=${MAX_LATENCY:-$MAX_LATENCY}
+PACKET_LOSS=${PACKET_LOSS:-$PACKET_LOSS}
+
+# Function to apply network configuration
+apply_tc_config() {
+    # Remove any existing qdisc configuration on eth0
+    tc qdisc del dev eth0 root 2>/dev/null
+    # Apply the new latency, jitter, and packet loss configuration
+    tc qdisc add dev eth0 root handle 1: netem delay ${CURRENT_LATENCY}ms ${JITTER}ms distribution normal loss ${PACKET_LOSS}%
+    # Apply the bandwidth limitation
+    tc qdisc add dev eth0 parent 1:1 handle 10: tbf rate ${BANDWIDTH}kbit burst 32kbit latency 50ms
+    echo "Updated configuration: Latency = ${CURRENT_LATENCY}ms, Jitter = ${JITTER}ms, Bandwidth = ${BANDWIDTH}kbit, Packet Loss = ${PACKET_LOSS}% - (distribution normal)"
+}
+
+# Initial configuration
+apply_tc_config
+echo "Initial configuration applied: Latency = ${CURRENT_LATENCY}ms, Jitter = ${JITTER}ms, Bandwidth = ${BANDWIDTH}kbit, Packet Loss = ${PACKET_LOSS}% - (distribution normal)"
+
+while true; do
+    sleep $INCREASE_TIME
+    CURRENT_LATENCY=$((CURRENT_LATENCY + 1000))
+    if [ $CURRENT_LATENCY -gt $MAX_LATENCY ]; then
+        CURRENT_LATENCY=$INITIAL_LATENCY
+    fi
+    apply_tc_config
+done

server/src/test/network-latency-simulator/setup.sh

@@ -0,0 +1,14 @@
+#!/bin/bash
+
+echo "Setting up test-context"
+
+mkdir -p test-context
+mkdir -p test-context/protos
+
+cp -R ../../../../protos/src/main/protobuf/*.proto test-context/protos/
+cp ../resources/consumer.sh test-context/
+
+# Make sure to make scripts executable
+chmod +x test-context/consumer.sh start.sh configure_latency.sh
+
+echo "Successfully set up test-context"

server/src/test/network-latency-simulator/start.sh

@@ -0,0 +1,18 @@
+# Print ENV Values
+echo "-----  Configuration of Consumer Variables:  -----"
+echo "GRPC_SERVER: $GRPC_SERVER"
+echo "GRPC_METHOD: $GRPC_METHOD"
+echo "PATH_TO_PROTO: $PATH_TO_PROTO"
+echo "PROTO_IMPORT_PATH: $PROTO_IMPORT_PATH"
+echo "-----  Configuration of Latency Variables:  -----"
+echo "INITIAL_LATENCY: $INITIAL_LATENCY"
+echo "JITTER: $JITTER"
+echo "BANDWIDTH: $BANDWIDTH"
+echo "INCREASE_TIME: $INCREASE_TIME"
+echo "MAX_LATENCY: $MAX_LATENCY"
+echo "PACKET_LOSS: $PACKET_LOSS"
+
+# First Start consumer without any network latency so it connects without issues.
+consumer.sh 1 1000 &
+# Then start the network latency configuration script.
+configure_latency.sh

server/src/test/resources/consumer.sh

@@ -21,9 +21,11 @@ if [ "$#" -eq 2 ]; then
   echo "The optional positive integer is: $2"
 fi
 
-GRPC_SERVER="localhost:8080"
-GRPC_METHOD="BlockStreamGrpc/StreamSource"
-PATH_TO_PROTO="../../../../protos/src/main/protobuf/blockstream.proto"
+# Use environment variables or default values
+GRPC_SERVER=${GRPC_SERVER:-"localhost:8080"}
+GRPC_METHOD=${GRPC_METHOD:-"BlockStreamGrpc/StreamSource"}
+PATH_TO_PROTO=${PATH_TO_PROTO:-"../../../../protos/src/main/protobuf/blockstream.proto"}
+PROTO_IMPORT_PATH=${PROTO_IMPORT_PATH:-"../../../../protos/src/main/protobuf"}
 
 echo "Starting consumer..."
 
@@ -54,5 +56,5 @@ trap cleanup SIGINT
     sleep 1
 
   done
-) | grpcurl -plaintext -proto $PATH_TO_PROTO -d @ $GRPC_SERVER $GRPC_METHOD
+) | grpcurl -plaintext -import-path $PROTO_IMPORT_PATH -proto $PATH_TO_PROTO -d @ $GRPC_SERVER $GRPC_METHOD