Add clusterUUID to flow table (antrea-io#108)

To support multi-cluster, we add a new column "clusterUUID" as the cluster identifier to flow table. This PR makes the corresponding changes in the ClickHouse dataschema and related tests. We found the ClickHouse init process is likely to fail due to timeout, so we increase the env value of CLICKHOUSE_INIT_TIMEOUT from 12s to 60s, as suggested by Altinity documentation: https://kb.altinity.com/altinity-kb-kubernetes/altinity-kb-possible-issues-with-running-clickhouse-in-k8s/#clickhouse-init-process-failed Signed-off-by: heanlan <hanlan@vmware.com>
heanlan · Sep 15, 2022 · c9ef257 · c9ef257
1 parent 673512b
commit c9ef257
Show file tree

Hide file tree

Showing 7 changed files with 78 additions and 24 deletions.
diff --git a/build/charts/theia/provisioning/datasources/create_table.sh b/build/charts/theia/provisioning/datasources/create_table.sh
@@ -77,6 +77,7 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
         throughputFromDestinationNode UInt64,
         reverseThroughputFromSourceNode UInt64,
         reverseThroughputFromDestinationNode UInt64,
+        clusterUUID String,
         trusted UInt8 DEFAULT 0
     ) engine=ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}')
     ORDER BY (timeInserted, flowEndSeconds)
@@ -100,7 +101,8 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
         sourcePodNamespace,
         destinationPodNamespace,
         sourceTransportPort,
-        destinationTransportPort)
+        destinationTransportPort,
+        clusterUUID)
     TTL timeInserted + INTERVAL {{ .Values.clickhouse.ttl }}
     SETTINGS merge_with_ttl_timeout = {{ $ttlTimeout }}
     POPULATE
@@ -124,7 +126,8 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
         sum(throughput) AS throughput,
         sum(reverseThroughput) AS reverseThroughput,
         sum(throughputFromSourceNode) AS throughputFromSourceNode,
-        sum(throughputFromDestinationNode) AS throughputFromDestinationNode
+        sum(throughputFromDestinationNode) AS throughputFromDestinationNode,
+        clusterUUID
     FROM flows_local
     GROUP BY
         timeInserted,
@@ -140,7 +143,8 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
         sourcePodNamespace,
         destinationPodNamespace,
         sourceTransportPort,
-        destinationTransportPort;
+        destinationTransportPort,
+        clusterUUID;
 
     --Create a Materialized View to aggregate data for nodes
     CREATE MATERIALIZED VIEW IF NOT EXISTS flows_node_view_local
@@ -153,7 +157,8 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
         sourceNodeName,
         destinationNodeName,
         sourcePodNamespace,
-        destinationPodNamespace)
+        destinationPodNamespace,
+        clusterUUID)
     TTL timeInserted + INTERVAL {{ .Values.clickhouse.ttl }}
     SETTINGS merge_with_ttl_timeout = {{ $ttlTimeout }}
     POPULATE
@@ -173,7 +178,8 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
         sum(throughputFromSourceNode) AS throughputFromSourceNode,
         sum(reverseThroughputFromSourceNode) AS reverseThroughputFromSourceNode,
         sum(throughputFromDestinationNode) AS throughputFromDestinationNode,
-        sum(reverseThroughputFromDestinationNode) AS reverseThroughputFromDestinationNode
+        sum(reverseThroughputFromDestinationNode) AS reverseThroughputFromDestinationNode,
+        clusterUUID
     FROM flows_local
     GROUP BY
         timeInserted,
@@ -183,7 +189,8 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
         sourceNodeName,
         destinationNodeName,
         sourcePodNamespace,
-        destinationPodNamespace;
+        destinationPodNamespace,
+        clusterUUID;
 
     --Create a Materialized View to aggregate data for network policies
     CREATE MATERIALIZED VIEW IF NOT EXISTS flows_policy_view_local
@@ -207,7 +214,8 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
         destinationPodNamespace,
         destinationServicePort,
         destinationServicePortName,
-        destinationIP)
+        destinationIP,
+        clusterUUID)
     TTL timeInserted + INTERVAL {{ .Values.clickhouse.ttl }}
     SETTINGS merge_with_ttl_timeout = {{ $ttlTimeout }}
     POPULATE
@@ -238,7 +246,8 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
         sum(throughputFromSourceNode) AS throughputFromSourceNode,
         sum(reverseThroughputFromSourceNode) AS reverseThroughputFromSourceNode,
         sum(throughputFromDestinationNode) AS throughputFromDestinationNode,
-        sum(reverseThroughputFromDestinationNode) AS reverseThroughputFromDestinationNode
+        sum(reverseThroughputFromDestinationNode) AS reverseThroughputFromDestinationNode,
+        clusterUUID
     FROM flows_local
     GROUP BY
         timeInserted,
@@ -259,7 +268,8 @@ clickhouse client -n -h 127.0.0.1 <<-EOSQL
         destinationPodNamespace,
         destinationServicePort,
         destinationServicePortName,
-        destinationIP;
+        destinationIP,
+        clusterUUID;
 
     --Create a table to store the network policy recommendation results
     CREATE TABLE IF NOT EXISTS recommendations_local (

diff --git a/build/charts/theia/provisioning/datasources/migrators/downgrade/0-3-0_0-2-0.sql b/build/charts/theia/provisioning/datasources/migrators/downgrade/0-3-0_0-2-0.sql
@@ -0,0 +1,5 @@
+--Alter table to drop new columns
+ALTER TABLE flows
+DROP clusterUUID String;
+ALTER TABLE flows_local
+DROP clusterUUID String;
diff --git a/build/charts/theia/provisioning/datasources/migrators/upgrade/0-2-0_0-3-0.sql b/build/charts/theia/provisioning/datasources/migrators/upgrade/0-2-0_0-3-0.sql
@@ -0,0 +1,5 @@
+--Alter table to add new columns
+ALTER TABLE flows
+ADD COLUMN clusterUUID String;
+ALTER TABLE flows_local
+ADD COLUMN clusterUUID String;
diff --git a/build/charts/theia/templates/_helpers.tpl b/build/charts/theia/templates/_helpers.tpl
@@ -50,6 +50,8 @@
   env:
     - name: THEIA_VERSION
       value: {{ $version }}
+    - name: CLICKHOUSE_INIT_TIMEOUT
+      value: "60"
 {{- end }}
 
 {{- define "clickhouse.volume" }}

diff --git a/build/yamls/flow-visibility.yml b/build/yamls/flow-visibility.yml
@@ -309,6 +309,18 @@ data:
     DROP TABLE recommendations;
   0-2-0_0-1-0.sql: |
     -- place holder
+  0-2-0_0-3-0.sql: |
+    --Alter table to add new columns
+    ALTER TABLE flows
+    ADD COLUMN clusterUUID String;
+    ALTER TABLE flows_local
+    ADD COLUMN clusterUUID String;
+  0-3-0_0-2-0.sql: |
+    --Alter table to drop new columns
+    ALTER TABLE flows
+    DROP clusterUUID String;
+    ALTER TABLE flows_local
+    DROP clusterUUID String;
   create_table.sh: |
     #!/usr/bin/env bash
 
@@ -380,6 +392,7 @@ data:
             throughputFromDestinationNode UInt64,
             reverseThroughputFromSourceNode UInt64,
             reverseThroughputFromDestinationNode UInt64,
+            clusterUUID String,
             trusted UInt8 DEFAULT 0
         ) engine=ReplicatedMergeTree('/clickhouse/tables/{shard}/{database}/{table}', '{replica}')
         ORDER BY (timeInserted, flowEndSeconds)
@@ -403,7 +416,8 @@ data:
             sourcePodNamespace,
             destinationPodNamespace,
             sourceTransportPort,
-            destinationTransportPort)
+            destinationTransportPort,
+            clusterUUID)
         TTL timeInserted + INTERVAL 12 HOUR
         SETTINGS merge_with_ttl_timeout = 14400
         POPULATE
@@ -427,7 +441,8 @@ data:
             sum(throughput) AS throughput,
             sum(reverseThroughput) AS reverseThroughput,
             sum(throughputFromSourceNode) AS throughputFromSourceNode,
-            sum(throughputFromDestinationNode) AS throughputFromDestinationNode
+            sum(throughputFromDestinationNode) AS throughputFromDestinationNode,
+            clusterUUID
         FROM flows_local
         GROUP BY
             timeInserted,
@@ -443,7 +458,8 @@ data:
             sourcePodNamespace,
             destinationPodNamespace,
             sourceTransportPort,
-            destinationTransportPort;
+            destinationTransportPort,
+            clusterUUID;
 
         --Create a Materialized View to aggregate data for nodes
         CREATE MATERIALIZED VIEW IF NOT EXISTS flows_node_view_local
@@ -456,7 +472,8 @@ data:
             sourceNodeName,
             destinationNodeName,
             sourcePodNamespace,
-            destinationPodNamespace)
+            destinationPodNamespace,
+            clusterUUID)
         TTL timeInserted + INTERVAL 12 HOUR
         SETTINGS merge_with_ttl_timeout = 14400
         POPULATE
@@ -476,7 +493,8 @@ data:
             sum(throughputFromSourceNode) AS throughputFromSourceNode,
             sum(reverseThroughputFromSourceNode) AS reverseThroughputFromSourceNode,
             sum(throughputFromDestinationNode) AS throughputFromDestinationNode,
-            sum(reverseThroughputFromDestinationNode) AS reverseThroughputFromDestinationNode
+            sum(reverseThroughputFromDestinationNode) AS reverseThroughputFromDestinationNode,
+            clusterUUID
         FROM flows_local
         GROUP BY
             timeInserted,
@@ -486,7 +504,8 @@ data:
             sourceNodeName,
             destinationNodeName,
             sourcePodNamespace,
-            destinationPodNamespace;
+            destinationPodNamespace,
+            clusterUUID;
 
         --Create a Materialized View to aggregate data for network policies
         CREATE MATERIALIZED VIEW IF NOT EXISTS flows_policy_view_local
@@ -510,7 +529,8 @@ data:
             destinationPodNamespace,
             destinationServicePort,
             destinationServicePortName,
-            destinationIP)
+            destinationIP,
+            clusterUUID)
         TTL timeInserted + INTERVAL 12 HOUR
         SETTINGS merge_with_ttl_timeout = 14400
         POPULATE
@@ -541,7 +561,8 @@ data:
             sum(throughputFromSourceNode) AS throughputFromSourceNode,
             sum(reverseThroughputFromSourceNode) AS reverseThroughputFromSourceNode,
             sum(throughputFromDestinationNode) AS throughputFromDestinationNode,
-            sum(reverseThroughputFromDestinationNode) AS reverseThroughputFromDestinationNode
+            sum(reverseThroughputFromDestinationNode) AS reverseThroughputFromDestinationNode,
+            clusterUUID
         FROM flows_local
         GROUP BY
             timeInserted,
@@ -562,7 +583,8 @@ data:
             destinationPodNamespace,
             destinationServicePort,
             destinationServicePortName,
-            destinationIP;
+            destinationIP,
+            clusterUUID;
 
         --Create a table to store the network policy recommendation results
         CREATE TABLE IF NOT EXISTS recommendations_local (
@@ -6534,6 +6556,8 @@ spec:
         - env:
           - name: THEIA_VERSION
             value: 0.3.0
+          - name: CLICKHOUSE_INIT_TIMEOUT
+            value: "60"
           image: projects.registry.vmware.com/antrea/theia-clickhouse-server:22.6
           imagePullPolicy: IfNotPresent
           name: clickhouse
@@ -6583,8 +6607,12 @@ spec:
               path: migrate.sh
             - key: 0-1-0_0-2-0.sql
               path: migrators/upgrade/0-1-0_0-2-0.sql
+            - key: 0-2-0_0-3-0.sql
+              path: migrators/upgrade/0-2-0_0-3-0.sql
             - key: 0-2-0_0-1-0.sql
               path: migrators/downgrade/0-2-0_0-1-0.sql
+            - key: 0-3-0_0-2-0.sql
+              path: migrators/downgrade/0-3-0_0-2-0.sql
             name: clickhouse-mounted-configmap
           name: clickhouse-configmap-volume
         - emptyDir:

diff --git a/test/e2e/flowvisibility_test.go b/test/e2e/flowvisibility_test.go
@@ -94,6 +94,7 @@ throughputFromSourceNode: 15902813472
 throughputFromDestinationNode: 15902813472
 reverseThroughputFromSourceNode: 12381344
 reverseThroughputFromDestinationNode: 12381344
+clusterUUID: 7e2e1de2-c85f-476e-ab1a-fce1bf83ee2c
 trusted: 0
 */
 
@@ -1251,6 +1252,7 @@ type ClickHouseFullRow struct {
 	ThroughputFromDestinationNode        uint64    `json:"throughputFromDestinationNode,string"`
 	ReverseThroughputFromSourceNode      uint64    `json:"reverseThroughputFromSourceNode,string"`
 	ReverseThroughputFromDestinationNode uint64    `json:"reverseThroughputFromDestinationNode,string"`
+	ClusterUUID                          string    `json:"clusterUUID"`
 	Trusted                              uint8     `json:"trusted"`
 }
 

diff --git a/test/e2e/theia_clickhouse_test.go b/test/e2e/theia_clickhouse_test.go
@@ -83,12 +83,13 @@ const (
                    throughputFromSourceNode,
                    throughputFromDestinationNode,
                    reverseThroughputFromSourceNode,
-                   reverseThroughputFromDestinationNode)
+                   reverseThroughputFromDestinationNode,
+                   clusterUUID)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                            ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                            ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
                            ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
-                           ?, ?, ?, ?, ?, ?, ?)`
+                           ?, ?, ?, ?, ?, ?, ?, ?)`
 	recordPerCommit      = 1000
 	insertInterval       = 1
 	threshold            = 25
@@ -101,10 +102,10 @@ const (
 )
 
 var tableColumnNumberMap = map[string]string{
-	".inner.flows_node_view_local":   "16",
-	".inner.flows_pod_view_local":    "20",
-	".inner.flows_policy_view_local": "27",
-	"flows_local":                    "49",
+	".inner.flows_node_view_local":   "17",
+	".inner.flows_pod_view_local":    "21",
+	".inner.flows_policy_view_local": "28",
+	"flows_local":                    "50",
 	"recommendations_local":          "4",
 }
 
@@ -339,6 +340,7 @@ func addFakeRecord(t *testing.T, stmt *sql.Stmt) {
 		uint64(randInt(t, MaxInt32)),
 		uint64(randInt(t, MaxInt32)),
 		uint64(randInt(t, MaxInt32)),
+		"",
 	)
 	require.NoError(t, err)
 }