support_lineage

Zzm0809 · Oct 9, 2023 · e3d27e1 · e3d27e1
1 parent bafa357
commit e3d27e1
Show file tree

Hide file tree

Showing 39 changed files with 1,315 additions and 570 deletions.
diff --git a/dinky-admin/src/main/java/org/dinky/data/dto/JobDataDto.java b/dinky-admin/src/main/java/org/dinky/data/dto/JobDataDto.java
@@ -19,10 +19,10 @@
 
 package org.dinky.data.dto;
 
+import org.dinky.data.flink.config.FlinkJobConfigInfo;
+import org.dinky.data.flink.exceptions.FlinkJobExceptionsDetail;
+import org.dinky.data.flink.job.FlinkJobDetailInfo;
 import org.dinky.data.model.JobHistory;
-import org.dinky.data.model.flink.config.FlinkJobConfigInfo;
-import org.dinky.data.model.flink.exceptions.FlinkJobExceptionsDetail;
-import org.dinky.data.model.flink.job.FlinkJobDetailInfo;
 import org.dinky.utils.JsonUtils;
 
 import java.time.LocalDateTime;

diff --git a/dinky-admin/src/main/java/org/dinky/job/handler/JobRefreshHandler.java b/dinky-admin/src/main/java/org/dinky/job/handler/JobRefreshHandler.java
@@ -26,13 +26,13 @@
 import org.dinky.data.dto.ClusterConfigurationDTO;
 import org.dinky.data.dto.JobDataDto;
 import org.dinky.data.enums.JobStatus;
+import org.dinky.data.flink.backpressure.FlinkJobNodeBackPressure;
+import org.dinky.data.flink.config.FlinkJobConfigInfo;
+import org.dinky.data.flink.exceptions.FlinkJobExceptionsDetail;
+import org.dinky.data.flink.job.FlinkJobDetailInfo;
+import org.dinky.data.flink.watermark.FlinkJobNodeWaterMark;
 import org.dinky.data.model.JobInfoDetail;
 import org.dinky.data.model.JobInstance;
-import org.dinky.data.model.flink.backpressure.FlinkJobNodeBackPressure;
-import org.dinky.data.model.flink.config.FlinkJobConfigInfo;
-import org.dinky.data.model.flink.exceptions.FlinkJobExceptionsDetail;
-import org.dinky.data.model.flink.job.FlinkJobDetailInfo;
-import org.dinky.data.model.flink.watermark.FlinkJobNodeWaterMark;
 import org.dinky.gateway.Gateway;
 import org.dinky.gateway.config.GatewayConfig;
 import org.dinky.gateway.enums.GatewayType;

diff --git a/dinky-admin/src/main/resources/db/db-h2.sql b/dinky-admin/src/main/resources/db/db-h2.sql
@@ -1338,7 +1338,7 @@ CREATE TABLE `dinky_task` (
                             `alert_group_id` bigint(20) null DEFAULT null COMMENT 'alert group id',
                             `config_json` text null COMMENT 'configuration json',
                             `note` varchar(255) null DEFAULT null COMMENT 'Job Note',
-                            `step` int(11) null DEFAULT null COMMENT 'Job lifecycle',
+                            `step` int(11) null DEFAULT 1 COMMENT 'Job lifecycle',
                             `job_instance_id` bigint(20) null DEFAULT null COMMENT 'job instance id',
                             `enabled` tinyint(1) NOT null DEFAULT 1 COMMENT 'is enable',
                             `create_time` datetime(0) null DEFAULT null COMMENT 'create time',

diff --git a/dinky-alert/dinky-alert-base/src/main/java/org/dinky/alert/Rules/ExceptionRule.java b/dinky-alert/dinky-alert-base/src/main/java/org/dinky/alert/Rules/ExceptionRule.java
@@ -19,9 +19,10 @@
 
 package org.dinky.alert.Rules;
 
+import org.dinky.data.flink.exceptions.FlinkJobExceptionsDetail;
+
 import java.util.concurrent.TimeUnit;
 
-import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
@@ -43,20 +44,20 @@ public ExceptionRule() {
      * @param exceptions The exceptions object containing relevant data.
      * @return True if the operation should be executed, false otherwise.
      */
-    public Boolean isException(Integer key, ObjectNode exceptions) {
+    public Boolean isException(Integer key, FlinkJobExceptionsDetail exceptions) {
 
         // If the exception is the same as the previous one, it will not be reported again
-        if (exceptions.get("timestamp") == null) {
+        if (exceptions.getTimestamp() == null) {
             return false;
         }
-        long timestamp = exceptions.get("timestamp").asLong(0);
+        long timestamp = exceptions.getTimestamp();
         Long hisTimeIfPresent = hisTime.getIfPresent(key);
         if (hisTimeIfPresent != null && hisTimeIfPresent == timestamp) {
             return false;
         }
         hisTime.put(key, timestamp);
-        if (exceptions.has("root-exception")) {
-            return !exceptions.get("root-exception").isNull();
+        if (exceptions.getRootException() != null) {
+            return !exceptions.getRootException().isEmpty();
         } else {
             return false;
         }

diff --git a/dinky-common/pom.xml b/dinky-common/pom.xml
@@ -51,7 +51,10 @@
             <groupId>cn.hutool</groupId>
             <artifactId>hutool-all</artifactId>
         </dependency>
-
+        <dependency>
+            <groupId>com.alibaba.fastjson2</groupId>
+            <artifactId>fastjson2</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.github.docker-java</groupId>
             <artifactId>docker-java-core</artifactId>

diff --git a/...ackpressure/FlinkJobNodeBackPressure.java → ...ackpressure/FlinkJobNodeBackPressure.java b/...ackpressure/FlinkJobNodeBackPressure.java → ...ackpressure/FlinkJobNodeBackPressure.java
@@ -17,7 +17,7 @@
  *
  */
 
-package org.dinky.data.model.flink.backpressure;
+package org.dinky.data.flink.backpressure;
 
 import java.io.Serializable;
 import java.util.List;

diff --git a/...a/model/flink/config/ExecutionConfig.java → ...ky/data/flink/config/ExecutionConfig.java b/...a/model/flink/config/ExecutionConfig.java → ...ky/data/flink/config/ExecutionConfig.java
@@ -17,7 +17,7 @@
  *
  */
 
-package org.dinky.data.model.flink.config;
+package org.dinky.data.flink.config;
 
 import java.io.Serializable;
 

diff --git a/...odel/flink/config/FlinkJobConfigInfo.java → ...data/flink/config/FlinkJobConfigInfo.java b/...odel/flink/config/FlinkJobConfigInfo.java → ...data/flink/config/FlinkJobConfigInfo.java
@@ -17,7 +17,7 @@
  *
  */
 
-package org.dinky.data.model.flink.config;
+package org.dinky.data.flink.config;
 
 import java.io.Serializable;
 

diff --git a/.../exceptions/FlinkJobExceptionsDetail.java → .../exceptions/FlinkJobExceptionsDetail.java b/.../exceptions/FlinkJobExceptionsDetail.java → .../exceptions/FlinkJobExceptionsDetail.java
@@ -17,7 +17,7 @@
  *
  */
 
-package org.dinky.data.model.flink.exceptions;
+package org.dinky.data.flink.exceptions;
 
 import java.io.Serializable;
 import java.util.List;

diff --git a/...a/model/flink/job/FlinkJobDetailInfo.java → ...ky/data/flink/job/FlinkJobDetailInfo.java b/...a/model/flink/job/FlinkJobDetailInfo.java → ...ky/data/flink/job/FlinkJobDetailInfo.java
@@ -17,7 +17,7 @@
  *
  */
 
-package org.dinky.data.model.flink.job;
+package org.dinky.data.flink.job;
 
 import java.io.Serializable;
 import java.util.List;

diff --git a/...ky/data/model/flink/job/FlinkJobPlan.java → ...rg/dinky/data/flink/job/FlinkJobPlan.java b/...ky/data/model/flink/job/FlinkJobPlan.java → ...rg/dinky/data/flink/job/FlinkJobPlan.java
@@ -17,7 +17,7 @@
  *
  */
 
-package org.dinky.data.model.flink.job;
+package org.dinky.data.flink.job;
 
 import java.io.Serializable;
 import java.util.List;

diff --git a/...ata/model/flink/job/FlinkJobPlanNode.java → ...inky/data/flink/job/FlinkJobPlanNode.java b/...ata/model/flink/job/FlinkJobPlanNode.java → ...inky/data/flink/job/FlinkJobPlanNode.java
@@ -17,10 +17,10 @@
  *
  */
 
-package org.dinky.data.model.flink.job;
+package org.dinky.data.flink.job;
 
-import org.dinky.data.model.flink.backpressure.FlinkJobNodeBackPressure;
-import org.dinky.data.model.flink.watermark.FlinkJobNodeWaterMark;
+import org.dinky.data.flink.backpressure.FlinkJobNodeBackPressure;
+import org.dinky.data.flink.watermark.FlinkJobNodeWaterMark;
 
 import java.io.Serializable;
 import java.util.List;

diff --git a/...odel/flink/job/FlinkJobPlanNodeInput.java → ...data/flink/job/FlinkJobPlanNodeInput.java b/...odel/flink/job/FlinkJobPlanNodeInput.java → ...data/flink/job/FlinkJobPlanNodeInput.java
@@ -17,7 +17,7 @@
  *
  */
 
-package org.dinky.data.model.flink.job;
+package org.dinky.data.flink.job;
 
 import java.io.Serializable;
 

diff --git a/.../data/model/flink/job/FlinkJobVertex.java → .../dinky/data/flink/job/FlinkJobVertex.java b/.../data/model/flink/job/FlinkJobVertex.java → .../dinky/data/flink/job/FlinkJobVertex.java
@@ -17,7 +17,7 @@
  *
  */
 
-package org.dinky.data.model.flink.job;
+package org.dinky.data.flink.job;
 
 import java.io.Serializable;
 import java.util.Map;

diff --git a/...link/watermark/FlinkJobNodeWaterMark.java → ...link/watermark/FlinkJobNodeWaterMark.java b/...link/watermark/FlinkJobNodeWaterMark.java → ...link/watermark/FlinkJobNodeWaterMark.java
@@ -17,7 +17,7 @@
  *
  */
 
-package org.dinky.data.model.flink.watermark;
+package org.dinky.data.flink.watermark;
 
 import java.io.Serializable;
 

diff --git a/dinky-web/package.json b/dinky-web/package.json
@@ -62,6 +62,7 @@
     "react-countup": "^6.4.2",
     "react-dom": "^18.0.0",
     "react-helmet-async": "^1.3.0",
+    "react-lineage-dag": "^2.0.36",
     "react-markdown": "^8.0.7",
     "react-spring": "^9.7.3",
     "react-use-cookie": "^1.4.0",

diff --git a/dinky-web/src/components/CallBackButton/CircleBtn.tsx b/dinky-web/src/components/CallBackButton/CircleBtn.tsx
@@ -16,9 +16,9 @@
  *
  */
 
+import { TabsItemType } from '@/pages/DataStudio/model';
 import { Button } from 'antd';
 import React from 'react';
-import {TabsItemType} from "@/pages/DataStudio/model";
 
 export type CircleButtonProps = {
   icon: React.ReactNode;
@@ -30,7 +30,7 @@ export type CircleButtonProps = {
 export type CircleDataStudioButtonProps = {
   icon: React.ReactNode;
   loading?: boolean;
-  onClick?:  (panes:TabsItemType[],activeKey:string) => void;
+  onClick?: (panes: TabsItemType[], activeKey: string) => void;
   title?: string;
   key?: string;
 };

diff --git a/dinky-web/src/components/FlinkDag/component/DagDataNode.tsx b/dinky-web/src/components/FlinkDag/component/DagDataNode.tsx
@@ -98,13 +98,20 @@ const DagDataNode = (props: any) => {
           <Text style={{ display: 'inline-flex', alignItems: 'center' }} type='secondary'>
             {' '}
             {l('devops.baseinfo.busy')}:
-            {renderRatio((backpressure && backpressure.subtasks)?backpressure.subtasks[0]?.busyRatio:0, false)}
+            {renderRatio(
+              backpressure && backpressure.subtasks ? backpressure.subtasks[0]?.busyRatio : 0,
+              false
+            )}
           </Text>
         </Col>
         <Col flex='auto'>
           <Text type='secondary' ellipsis>
             {l('devops.baseinfo.backpressure')}:
-            <StatusTag status={backpressure?backpressure.status:0} bordered={false} animation={false} />
+            <StatusTag
+              status={backpressure ? backpressure.status : 0}
+              bordered={false}
+              animation={false}
+            />
           </Text>
         </Col>
       </Row>
@@ -113,7 +120,10 @@ const DagDataNode = (props: any) => {
         <Col flex='35%'>
           <Text style={{ display: 'inline-flex', alignItems: 'center' }} type='secondary'>
             {l('devops.baseinfo.idle')}:
-            {renderRatio((backpressure && backpressure.subtasks)?backpressure.subtasks[0]?.idleRatio:0, true)}
+            {renderRatio(
+              backpressure && backpressure.subtasks ? backpressure.subtasks[0]?.idleRatio : 0,
+              true
+            )}
           </Text>
         </Col>
         <Col flex='auto'>
-Original file line number
+Diff line change
@@ Expand Up / @@ -17,7 +17,7 @@ @@
      *
      */
-    package org.dinky.data.model.flink.config;
+    package org.dinky.data.flink.config;
     import java.io.Serializable;
@@ Expand Down @@