enh: change method to cal insert perf

Author: jiajingbin

.github/workflows/ems-test.yml

@@ -425,10 +425,11 @@ jobs:
       group: customers
       labels: ${{ matrix.runner }}
     steps:
-      - name: Export Client IP
+      - name: Export Client Host
         run: |
-          ip=$(hostname -I | awk '{print $1}')
-          echo "CLIENT_IP=$ip" >> $GITHUB_ENV
+          # ip=$(hostname -I | awk '{print $1}')
+          # echo "CLIENT_IP=$ip" >> $GITHUB_ENV
+          echo "CLIENT_HOST=$HOSTNAME" >> $GITHUB_ENV
 
       # - name: Sync Repo - taos-test-framework
       #   uses: taosdata/.github/.github/actions/sync-repo@main
@@ -512,8 +513,8 @@ jobs:
       - name: Import Process Exporter Dashboard
         uses: taosdata/.github/.github/actions/import-process-exporter-dashboard@main
         with:
-          grafana-url: http://${{ env.CLIENT_IP }}:3000
-          prometheus-url: http://${{ env.CLIENT_IP }}:9090
+          grafana-url: http://${{ env.CLIENT_HOST }}:3000
+          prometheus-url: http://${{ env.CLIENT_HOST }}:9090
           datasource-name: ${{ env.PROCESS_EXPORTER_DATASOURCE_NAME }}
 
       - name: Install Grafana Plugin
@@ -525,7 +526,7 @@ jobs:
       - name: Import Grafana Dashboard
         uses: taosdata/.github/.github/actions/import-grafana-dashboard@main
         with:
-          grafana-url: http://${{ env.CLIENT_IP }}:3000
+          grafana-url: http://${{ env.CLIENT_HOST }}:3000
           dashboard-ids: ${{ env.TDINSIGHT_DASHBOARD_IDS }}
           dashboard-uids: ${{ env.TDINSIGHT_DASHBOARD_UIDS }}
 
@@ -553,7 +554,7 @@ jobs:
           EDGE_DNODE_HOSTS=$SINGLE_DNODE_HOSTS
           TAOS_BENCHMARK_HOSTS=$TAOS_BENCHMARK_HOSTS
           CENTER_DNODE_HOSTS=$CLUSTER_HOSTS
-          GRAFANA_IP="${{ env.CLIENT_IP }}"
+          GRAFANA_IP="${{ env.CLIENT_HOST }}"
           GRAFANA_DATASOURCE_NAME=$(echo "${{ env.TDINSIGHT_DASHBOARD_UIDS }}" | cut -d ',' -f 1)
 
           # Combine Inputs into a single JSON object

tests/cases/customer_scenarios/ems/ems_edge.py

@@ -5,6 +5,7 @@
 from taostest.util.rest import TDRest
 from taostest.util import file
 import sys
+import time
 class EMSEdge(TDCase):
     def init(self):
         self.yml_name = sys.argv[1].split("=")[1]
@@ -64,6 +65,11 @@ def run(self):
             task_info = response.json()
             task_list.append(task_info["id"])
 
+        task_start_time = time.time()
+        self.case_config["task_start_time"] = task_start_time
+        with open(os.path.join(self.env_root, "workflow_config.json"), "w") as config_file:
+            json.dump(self.case_config, config_file, indent=4)
+
         # time.sleep(self.execute_time)
 
         # for task_id in task_list:

tests/cases/customer_scenarios/ems/ems_summary.py

@@ -35,6 +35,7 @@ def init(self):
         self.taosd_url = f'http://{self.center_first_ep_host}:6041/rest/sql'
         self.taosd_headers = {"Authorization": "Basic cm9vdDp0YW9zZGF0YQ=="}
         self.mqtt_received_bytes = 0
+        self.task_run_time = int(self.case_config["exec_time"]) - (self.case_config["task_start_time"] - self.case_config["mqtt_start_time"])
         self.test_robot_url = (
     "https://open.feishu.cn/open-apis/bot/v2/hook/11e9e452-34a0-4c88-b014-10e21cb521dd"
 )
@@ -95,6 +96,7 @@ def validate_sync(self):
         stable_counter = 0
 
         while time.time() - start_time < self.retention_timeout:
+            edge_total = self.collect_edge_data()
             current_center = self._get_center_data()
             self._remote._logger.info(f'current edge data: {edge_total}')
             self._remote._logger.info(f'current center data: {current_center}')
@@ -119,6 +121,15 @@ def validate_sync(self):
         ratio = final_center / edge_total if edge_total > 0 else 0
         return [f"{round(ratio*100, 2)}%", final_center, edge_total]
 
+    def write_final_edge_perf(self, insert_perf):
+        for node, perf_info in list(zip(self.edge_host_list, insert_perf)):
+            if perf_info["host"] == node:
+                perf_info["total_written_rows"] = self.stable_data[node]
+                perf_info["total_rows_per_second"] = round(self.stable_data[node]/self.task_run_time, 2)
+                perf_info["total_points_per_second"] = round(perf_info["total_written_rows"]/self.task_run_time, 2)
+                perf_info["total_rows_per_second"] = self.stable_data[node]
+        return perf_info
+
     def _get_center_data(self) -> int:
         stables = self._get_stables(self.center_first_ep_host, self.center_dbname)
         return sum(self._get_table_count(self.center_first_ep_host, stable, self.center_dbname) for stable in stables)
@@ -288,6 +299,7 @@ def run(self):
         data_retention_info["center_total_rows"] = center_total_rows
         data_retention_info["edge_total_rows"] = edge_total_rows
         insert_perf = self.get_insert_result()
+        insert_perf_info = self.write_final_edge_perf(insert_perf)
         query_perf = self.get_query_detail_result()
         compression_ratio_disk_info = self.get_compression_ratio()
         compression_data_size = self._get_compression_data()
@@ -296,7 +308,7 @@ def run(self):
         test_specs = self.get_test_specs()
         final_res_dict = {
             "Test Specs": test_specs,
-            "Insert Performance": insert_perf,
+            "Insert Performance": insert_perf_info,
             "Query Performance": query_perf,
             "Compression Ratio": compression_ratio,
             "Data Retention Info": data_retention_info,

tests/cases/customer_scenarios/ems/get_metrics.py

@@ -20,7 +20,7 @@
 from datetime import datetime, timedelta
 from taostest.util.rest import TDRest
 
-class Stop(TDCase):
+class GetMetrics(TDCase):
     def init(self):
         self.tdCom = TDCom(self.tdSql)
         self._remote: Remote = Remote(self.logger)
@@ -73,10 +73,11 @@ def run(self) -> bool:
         query_summary_metrics = {
             "role": self.get_role(),
             "host": self.host,
-            "total_rows_per_second":0,
-            "total_points_per_second":0,
-            "total_written_rows":0,
-            "total_written_points":0,
+            "total_rows_per_second": 0,
+            "total_points_per_second": 0,
+            "total_written_rows": 0,
+            "total_written_points": 0,
+            "total_written_points": 0,
             "mqtt_received_bytes": 0
         }
         tmq_summary_metrics = {
@@ -96,10 +97,7 @@ def run(self) -> bool:
             # query_summary_metrics["total_inserted_sqls"] += metrics_dict[task_id]["total"]["total_inserted_sqls"]
             if "total_points_per_second" in metrics_dict[task_id]["total"]:
                 self.api_type = 0
-                query_summary_metrics["total_points_per_second"] += round(metrics_dict[task_id]["total"]["total_points_per_second"], 2)
                 query_summary_metrics["total_written_points"] += round(metrics_dict[task_id]["total"]["total_written_points"], 2)
-                query_summary_metrics["total_written_rows"] += round(metrics_dict[task_id]["total"]["total_written_rows"], 2)
-                query_summary_metrics["total_rows_per_second"] += round(metrics_dict[task_id]["total"]["total_rows_per_second"], 2)
                 if "mqtt_received_bytes" in metrics_dict[task_id]["current"]:
                     query_summary_metrics["mqtt_received_bytes"] += round(metrics_dict[task_id]["current"]["mqtt_received_bytes"], 2)
                 else:

tests/cases/customer_scenarios/ems/start.py

@@ -20,6 +20,7 @@
 import json
 from pathlib import Path
 import toml
+import time
 class Start(TDCase):
     def init(self):
         # mqtt_data_source: $mqttDataSource,
@@ -93,7 +94,10 @@ def start_battery_storage_datain(self):
             for mqtt_toml in self.toml_file_list:
                 self._remote.put(mqtt_host, mqtt_toml, os.path.dirname(mqtt_toml))
                 cmd_list.append(f"screen -L -Logfile /var/log/taos/{Path(mqtt_toml).stem}.log -d -m mqtt_pub --csv-file /opt/battery_storage_data.csv --csv-header topic,payload,qos,a,b,c --schema {mqtt_toml} --host {edge_host} --interval {mqtt_pub_interval}ms --exec-duration {exec_time}s")
-            self._remote.cmd(mqtt_host, cmd_list)
+            mqtt_start_time = time.time()
+            self.case_config["mqtt_start_time"] = mqtt_start_time
+            with open(os.path.join(self.env_root, "workflow_config.json"), "w") as config_file:
+                json.dump(self.case_config, config_file, indent=4)
 
     def start_taosx_service(self,host):
         self._remote.cmd(host,"systemctl stop taos-explorer")