feat: 实验实例运行状态改为 SSE 请求

react-ui/package.json

@@ -32,7 +32,7 @@
     "record": "cross-env NODE_ENV=development REACT_APP_ENV=test max record --scene=login",
     "serve": "umi-serve",
     "start": "cross-env UMI_ENV=dev max dev",
-    "start:dev": "cross-env REACT_APP_ENV=dev MOCK=none UMI_ENV=dev max dev",
+    "start:dev": "cross-env REACT_APP_ENV=dev MOCK=none UMI_ENV=dev UMI_DEV_SERVER_COMPRESS=none max dev",
     "start:mock": "cross-env REACT_APP_ENV=dev UMI_ENV=dev max dev",
     "start:pre": "cross-env REACT_APP_ENV=pre UMI_ENV=dev max dev",
     "start:test": "cross-env REACT_APP_ENV=test MOCK=none UMI_ENV=dev max dev",

react-ui/src/pages/Experiment/Info/index.jsx

@@ -28,6 +28,7 @@ function ExperimentText() {
   const [propsDrawerOpen, openPropsDrawer, closePropsDrawer, propsDrawerOpenRef] =
     useVisible(false);
   const navigate = useNavigate();
+  const evtSourceRef = useRef();
   const width = 110;
   const height = 36;
 
@@ -48,6 +49,9 @@ function ExperimentText() {
       if (timerRef.current) {
         clearTimeout(timerRef.current);
       }
+      if (evtSourceRef.current) {
+        evtSourceRef.current.close();
+      }
     };
   }, []);
 
@@ -68,7 +72,7 @@ function ExperimentText() {
           item.imgName = item.img.slice(0, item.img.length - 4);
         });
         workflowRef.current = dag;
-        getExperimentInstance(true);
+        getExperimentInstance();
       } catch (error) {
         // JSON.parse 错误
         console.log(error);
@@ -77,50 +81,30 @@ function ExperimentText() {
   };
 
   // 获取实验实例
-  const getExperimentInstance = async (first) => {
+  const getExperimentInstance = async () => {
     const [res] = await to(getExperimentIns(locationParams.id));
     if (res && res.data && workflowRef.current) {
       setExperimentIns(res.data);
-      const { status, nodes_status } = res.data;
+      const { status, nodes_status, argo_ins_ns, argo_ins_name } = res.data;
       const workflowData = workflowRef.current;
       const experimentStatusObjs = JSON.parse(nodes_status);
       workflowData.nodes.forEach((item) => {
-        const experimentNode = experimentStatusObjs?.[item.id] ?? {};
-        const { finishedAt, startedAt, phase, id } = experimentNode;
-        item.experimentStartTime = startedAt;
-        item.experimentEndTime = finishedAt;
-        item.experimentStatus = phase;
-        item.workflowId = id;
-        item.img = phase ? `${item.imgName}-${phase}.png` : `${item.imgName}.png`;
+        const experimentNode = experimentStatusObjs?.[item.id];
+        updateWorkflowNode(item, experimentNode);
       });
 
-      // 更新打开的抽屉数据
-      if (propsDrawerOpenRef.current && experimentNodeDataRef.current) {
-        const currentId = experimentNodeDataRef.current.id;
-        const node = workflowData.nodes.find((item) => item.id === currentId);
-        if (node) {
-          setExperimentNodeData(node);
-        }
-      }
-
-      getGraphData(workflowData, first);
-
-      // 运行中或者等待中，每5秒获取一次实验实例
-      if (status === ExperimentStatus.Pending || status === ExperimentStatus.Running) {
-        timerRef.current = setTimeout(() => {
-          getExperimentInstance(false);
-        }, 5 * 1000);
-      }
+      // 绘制图
+      getGraphData(workflowData, true);
 
       // 如果状态是 Pending, 打开第一个节点
       // 如果状态是 Running，打开第一个运行中的节点，如果没有运行中的节点，打开第一个节点
-      if (first && status === ExperimentStatus.Pending) {
+      if (status === ExperimentStatus.Pending) {
         const node = workflowData.nodes[0];
         if (node) {
           setExperimentNodeData(node);
           openPropsDrawer();
         }
-      } else if (first && status === ExperimentStatus.Running) {
+      } else if (status === ExperimentStatus.Running) {
         const node =
           workflowData.nodes.find((item) => item.experimentStatus === ExperimentStatus.Running) ??
           workflowData.nodes[0];
@@ -129,9 +113,81 @@ function ExperimentText() {
           openPropsDrawer();
         }
       }
+
+      // 运行中或者等待中，开启 SSE
+      if (status === ExperimentStatus.Pending || status === ExperimentStatus.Running) {
+        setupSSE(argo_ins_name, argo_ins_ns);
+      }
     }
   };
 
+  const setupSSE = (name, namespace) => {
+    const { origin } = location;
+    const evtSource = new EventSource(
+      `${origin}/api/v1/realtimeStatus?listOptions.fieldSelector=metadata.namespace%3D${namespace}%2Cmetadata.name%3D${name}`,
+      { withCredentials: true },
+    );
+    evtSource.onmessage = (event) => {
+      const data = event?.data;
+      if (!data) {
+        return;
+      }
+      try {
+        const dataJson = JSON.parse(data);
+        const statusData = dataJson?.result?.object?.status;
+        if (!statusData) {
+          return;
+        }
+        const { startedAt, finishedAt, phase, nodes = {} } = statusData;
+        setExperimentIns((prev) => ({
+          ...prev,
+          finish_time: finishedAt,
+          status: phase,
+        }));
+
+        const workflowData = workflowRef.current;
+        workflowData.nodes.forEach((item) => {
+          const experimentNode = Object.values(nodes).find((node) => node.displayName === item.id);
+          updateWorkflowNode(item, experimentNode);
+        });
+        getGraphData(workflowData, false);
+
+        // 更新打开的抽屉数据
+        if (propsDrawerOpenRef.current && experimentNodeDataRef.current) {
+          const currentId = experimentNodeDataRef.current.id;
+          const node = workflowData.nodes.find((item) => item.id === currentId);
+          if (node) {
+            setExperimentNodeData(node);
+          }
+        }
+        if (phase !== ExperimentStatus.Pending && phase !== ExperimentStatus.Running) {
+          evtSource.close();
+        }
+      } catch (error) {
+        console.log(error);
+      }
+    };
+    evtSource.onerror = (error) => {
+      console.log('sse error', error);
+    };
+
+    evtSourceRef.current = evtSource;
+  };
+
+  function updateWorkflowNode(workflowNode, statusNode) {
+    if (!statusNode) {
+      return;
+    }
+    const { finishedAt, startedAt, phase, id } = statusNode;
+    workflowNode.experimentStartTime = startedAt;
+    workflowNode.experimentEndTime = finishedAt;
+    workflowNode.experimentStatus = phase;
+    workflowNode.workflowId = id;
+    workflowNode.img = phase
+      ? `${workflowNode.imgName}-${phase}.png`
+      : `${workflowNode.imgName}.png`;
+  }
+
   // 根据数据，渲染图
   const getGraphData = (data, first) => {
     if (graph) {
@@ -151,7 +207,7 @@ function ExperimentText() {
       }
     } else {
       setTimeout(() => {
-        getGraphData(data);
+        getGraphData(data, first);
       }, 500);
     }
   };

react-ui/src/pages/Pipeline/editPipeline/index.jsx

@@ -18,7 +18,7 @@ import { findAllParentNodes } from './utils';
 let graph = null;
 
 const EditPipeline = () => {
-  const navgite = useNavigate();
+  const navigate = useNavigate();
   const locationParams = useParams(); //新版本获取路由参数接口
   const graphRef = useRef();
   const paramsDrawerRef = useRef();
@@ -104,9 +104,7 @@ const EditPipeline = () => {
     setTimeout(() => {
       const data = graph.save();
       console.log(data);
-      const errorNode = data.nodes.find((item) => {
-        return item.formError === true;
-      });
+      const errorNode = data.nodes.find((item) => item.formError === true);
       if (errorNode) {
         message.error(`【${errorNode.label}】节点必填项必须配置`);
         const graphNode = graph.findById(errorNode.id);
@@ -124,7 +122,7 @@ const EditPipeline = () => {
         message.success('保存成功');
         setTimeout(() => {
           if (val) {
-            navgite({ pathname: `/pipeline/template` });
+            navigate({ pathname: `/pipeline/template` });
           }
         }, 500);
       });