Add ClickHouse monitor e2e test (#66)

Signed-off-by: Yanjun Zhou <[email protected]>

Author: yanjunz97

Diff for: ci/jenkins/test-vmc.sh

@@ -335,7 +335,7 @@ function deliver_antrea {
 
     control_plane_ip="$(kubectl get nodes -o wide --no-headers=true | awk -v role="$CONTROL_PLANE_NODE_ROLE" '$3 ~ role {print $6}')"
 
-    ${GIT_CHECKOUT_DIR}/hack/generate-manifest.sh --no-grafana > ${GIT_CHECKOUT_DIR}/build/yamls/flow-visibility.yml
+    ${GIT_CHECKOUT_DIR}/hack/generate-manifest.sh --no-grafana --ch-size 100Mi --ch-monitor-threshold 0.1 > ${GIT_CHECKOUT_DIR}/build/yamls/flow-visibility.yml
     ${GIT_CHECKOUT_DIR}/hack/generate-manifest.sh --no-grafana --spark-operator > ${GIT_CHECKOUT_DIR}/build/yamls/flow-visibility-with-spark.yml
 
     ${SCP_WITH_ANTREA_CI_KEY} $GIT_CHECKOUT_DIR/build/charts/theia/crds/clickhouse-operator-install-bundle.yaml capv@${control_plane_ip}:~

Diff for: ci/kind/test-e2e-kind.sh

@@ -39,7 +39,7 @@ function print_usage {
 TESTBED_CMD=$(dirname $0)"/kind-setup.sh"
 YML_DIR=$(dirname $0)"/../../build/yamls"
 FLOW_AGGREGATOR_CMD=$(dirname $0)"/../../hack/generate-manifest-flow-aggregator.sh"
-FLOW_VISIBILITY_CMD=$(dirname $0)"/../../hack/generate-manifest.sh --no-grafana"
+FLOW_VISIBILITY_CMD=$(dirname $0)"/../../hack/generate-manifest.sh --no-grafana --ch-size 100Mi --ch-monitor-threshold 0.1"
 FLOW_VISIBILITY_WITH_SPARK_CMD=$(dirname $0)"/../../hack/generate-manifest.sh --no-grafana --spark-operator"
 CH_OPERATOR_YML=$(dirname $0)"/../../build/charts/theia/crds/clickhouse-operator-install-bundle.yaml"
 

Diff for: hack/generate-manifest.sh

@@ -23,9 +23,15 @@ function echoerr {
 _usage="Usage: $0 [--mode (dev|release)] [--keep] [--help|-h]
 Generate a YAML manifest for the Clickhouse-Grafana Flow-visibility Solution, using Helm and
 Kustomize, and print it to stdout.
-        --mode (dev|release)  Choose the configuration variant that you need (default is 'dev')
-        --spark-operator      Generate a manifest with Spark Operator enabled.
-        --no-grafana          Generate a manifest with Grafana disabled.
+        --mode (dev|release)                Choose the configuration variant that you need (default is 'dev')
+        --spark-operator                    Generate a manifest with Spark Operator enabled.
+        --no-grafana                        Generate a manifest with Grafana disabled.
+        --ch-size <size>                    Deploy the ClickHouse with a specific storage size. Can be a 
+                                            plain integer or as a fixed-point number using one of these quantity
+                                            suffixes: E, P, T, G, M, K. Or the power-of-two equivalents:
+                                            Ei, Pi, Ti, Gi, Mi, Ki. (default is 8Gi)
+        --ch-monitor-threshold <threshold>  Deploy the ClickHouse monitor with a specific threshold. Can
+                                            vary from 0 to 1. (default is 0.5)
 This tool uses Helm 3 (https://helm.sh/) and Kustomize (https://github.com/kubernetes-sigs/kustomize)
 to generate manifests for Theia. You can set the HELM and KUSTOMIZE environment variable to
 the path of the helm and kustomize binary you want us to use. Otherwise we will download the
@@ -43,6 +49,8 @@ function print_help {
 MODE="dev"
 SPARK_OP=false
 GRAFANA=true
+CH_SIZE="8Gi"
+CH_THRESHOLD=0.5
 
 while [[ $# -gt 0 ]]
 do
@@ -61,6 +69,14 @@ case $key in
     GRAFANA=false
     shift 1
     ;;
+    --ch-size)
+    CH_SIZE="$2"
+    shift 2
+    ;;
+    --ch-monitor-threshold)
+    CH_THRESHOLD="$2"
+    shift 2
+    ;;
     -h|--help)
     print_usage
     exit 0
@@ -114,6 +130,8 @@ fi
 
 HELM_VALUES=()
 
+HELM_VALUES+=("clickhouse.storage.size=$CH_SIZE" "clickhouse.monitor.threshold=$CH_THRESHOLD")
+
 if [ "$MODE" == "dev" ] && [ -n "$IMG_NAME" ]; then
     HELM_VALUES+=("clickhouse.monitorImage.repository=$IMG_NAME")
 fi

Diff for: test/e2e/flowvisibility_test.go

@@ -17,7 +17,6 @@ package e2e
 import (
 	"encoding/json"
 	"fmt"
-
 	"net"
 	"strconv"
 	"strings"
@@ -113,6 +112,9 @@ const (
 	antreaIngressTableInitFlowCount = 6
 	ingressTableInitFlowCount       = 1
 	egressTableInitFlowCount        = 1
+	monitorThreshold                = 0.1
+	monitorDeletePercentage         = 0.5
+	monitorIperfRounds              = 80
 )
 
 var (
@@ -462,6 +464,91 @@ func testHelper(t *testing.T, data *TestData, podAIPs, podBIPs, podCIPs, podDIPs
 			checkRecordsForFlows(t, data, podAIPs.ipv4.String(), svcC.Spec.ClusterIP, isServiceIPv6, false, true, false, false)
 		}
 	})
+
+	// ClickHouseMonitor tests ensure ClickHouse monitor inspects the ClickHouse
+	// Pod storage usage and deletes the data when the stoage usage grows above
+	// the threshold.
+	t.Run("ClickHouseMonitor", func(t *testing.T) {
+		var flow testFlow
+		if !isIPv6 {
+			flow = testFlow{
+				srcIP:      podAIPs.ipv4.String(),
+				dstIP:      podBIPs.ipv4.String(),
+				srcPodName: "perftest-a",
+				dstPodName: "perftest-b",
+			}
+		} else {
+			flow = testFlow{
+				srcIP:      podAIPs.ipv6.String(),
+				dstIP:      podBIPs.ipv6.String(),
+				srcPodName: "perftest-a",
+				dstPodName: "perftest-b",
+			}
+		}
+		checkClickHouseMonitor(t, data, isIPv6, flow)
+	})
+}
+
+func checkClickHouseMonitor(t *testing.T, data *TestData, isIPv6 bool, flow testFlow) {
+	checkClickHouseMonitorLogs(t, data, false, 0)
+	var cmdStr string
+	// iperf3 has a limit on maximum parallel streams at 128
+	if !isIPv6 {
+		cmdStr = fmt.Sprintf("iperf3 -u -c %s -P 128 -n 1", flow.dstIP)
+	} else {
+		cmdStr = fmt.Sprintf("iperf3 -u -6 -c %s -P 128 -n 1", flow.dstIP)
+	}
+	log.Infof("Generating flow records to exceed monitor threshold...")
+	for i := 0; i < monitorIperfRounds; i++ {
+		stdout, stderr, err := data.RunCommandFromPod(testNamespace, flow.srcPodName, "perftool", []string{"bash", "-c", cmdStr})
+		require.NoErrorf(t, err, "Error when running iPerf3 client: %v,\nstdout:%s\nstderr:%s", err, stdout, stderr)
+	}
+	log.Infof("Waiting for the flows to be exported...")
+	time.Sleep(30 * time.Second)
+	// Get the number of records in database before the monitor deletes the records
+	stdout, stderr, err := data.RunCommandFromPod(flowVisibilityNamespace, clickHousePodName, "clickhouse", []string{"bash", "-c", "clickhouse client -q \"SELECT COUNT() FROM default.flows\""})
+	require.NoErrorf(t, err, "Error when querying ClickHouse server: %v,\nstdout:%s\nstderr:%s", err, stdout, stderr)
+	numRecord, err := strconv.ParseInt(strings.TrimSuffix(stdout, "\n"), 10, 64)
+	require.NoErrorf(t, err, "Failed when parsing the number of records %v", err)
+	log.Infof("Waiting for the monitor to detect and clean up the ClickHouse storage")
+	time.Sleep(2 * time.Minute)
+	checkClickHouseMonitorLogs(t, data, true, numRecord)
+}
+
+func checkClickHouseMonitorLogs(t *testing.T, data *TestData, deleted bool, numRecord int64) {
+	logString, err := data.GetPodLogs(flowVisibilityNamespace, clickHousePodName,
+		&corev1.PodLogOptions{
+			Container: "clickhouse-monitor",
+		},
+	)
+	require.NoErrorf(t, err, "Error when getting ClickHouse monitor logs: %v", err)
+	// ClickHouse monitor log sample
+	// "Memory usage" total=2062296555 used=42475 percentage=2.0595970980516912e-05
+	logs := strings.Split(logString, "Memory usage")
+	memoryUsageLog := strings.Split(logs[len(logs)-1], "\n")[0]
+	assert.Contains(t, memoryUsageLog, "total=")
+	assert.Contains(t, memoryUsageLog, "used=")
+	assert.Contains(t, memoryUsageLog, "percentage=")
+	percentage, err := strconv.ParseFloat(strings.Split(memoryUsageLog, "percentage=")[1], 64)
+	require.NoErrorf(t, err, "Failed when parsing the memory usage percentage %v", err)
+	if !deleted {
+		assert.LessOrEqual(t, percentage, monitorThreshold)
+	} else {
+		assert.Greater(t, percentage, monitorThreshold)
+		// Monitor deletes records from table flows and related MVs
+		assert.Contains(t, logString, "ALTER TABLE default.flows DELETE WHERE timeInserted < toDateTime", "Monitor should delete records from Table flows")
+		assert.Contains(t, logString, "ALTER TABLE default.flows_pod_view DELETE WHERE timeInserted < toDateTime", "Monitor should delete records from View flows_pod_view")
+		assert.Contains(t, logString, "ALTER TABLE default.flows_node_view DELETE WHERE timeInserted < toDateTime", "Monitor should delete records from View flows_node_view")
+		assert.Contains(t, logString, "ALTER TABLE default.flows_policy_view DELETE WHERE timeInserted < toDateTime", "Monitor should delete records from View flows_policy_view")
+		assert.Contains(t, logString, "Skip rounds after a successful deletion", "Monitor should skip rounds after a successful deletion")
+		require.Contains(t, logString, "SELECT timeInserted FROM default.flows LIMIT 1 OFFSET ", "Monitor should log the deletion SQL command")
+		deletedRecordLog := strings.Split(logString, "SELECT timeInserted FROM default.flows LIMIT 1 OFFSET ")[1]
+		deletedRecordLog = strings.Split(deletedRecordLog, "\n")[0]
+		numDeletedRecord, err := strconv.ParseInt(deletedRecordLog, 10, 64)
+		require.NoErrorf(t, err, "Failed when parsing the number of deleted records %v", err)
+
+		assert.InDeltaf(t, numDeletedRecord, float64(numRecord)*monitorDeletePercentage, float64(numDeletedRecord)*0.15, "Difference between expected and actual number of deleted Records should be lower than 15%")
+	}
 }
 
 func checkRecordsForFlows(t *testing.T, data *TestData, srcIP string, dstIP string, isIPv6 bool, isIntraNode bool, checkService bool, checkK8sNetworkPolicy bool, checkAntreaNetworkPolicy bool) {

Diff for: test/e2e/framework.go

@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"net"
 	"regexp"
 	"strconv"
@@ -649,6 +650,25 @@ func (data *TestData) PodWaitFor(timeout time.Duration, name, namespace string,
 	return pod, nil
 }
 
+// Gets pod logs from Pod
+func (data *TestData) GetPodLogs(namespace, name string, podLogOpts *corev1.PodLogOptions) (string, error) {
+	var logString string
+	req := data.clientset.CoreV1().Pods(namespace).GetLogs(name, podLogOpts)
+	podLogs, err := req.Stream(context.TODO())
+	if err != nil {
+		return logString, fmt.Errorf("error when opening stream: %v", err)
+	}
+	defer podLogs.Close()
+
+	buf := new(bytes.Buffer)
+	_, err = io.Copy(buf, podLogs)
+	if err != nil {
+		return logString, fmt.Errorf("error when copying data from Pod logs to buffer: %v", err)
+	}
+	logString = buf.String()
+	return logString, nil
+}
+
 func parsePodIPs(podIPStrings sets.String) (*PodIPs, error) {
 	ips := new(PodIPs)
 	for idx := range podIPStrings.List() {