Merge pull request #54 from aliyun/feature/release-v1.1.9-16

release 1.1.9-16

Author: Garfier

.github/workflows/main.yml

@@ -32,7 +32,7 @@ jobs:
       shell: bash
     - name: Archive test results
       if: always()
-      uses: actions/upload-artifact@v3
+      uses: actions/upload-artifact@v4
       with:
         name: test-results
         path: |

client/client-common/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>com.aliyun.dataworks</groupId>
         <artifactId>client</artifactId>
-        <version>1.1.9-10</version>
+        <version>1.1.9-16</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

client/client-spark-utils/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>com.aliyun.dataworks</groupId>
         <artifactId>client</artifactId>
-        <version>1.1.9-10</version>
+        <version>1.1.9-16</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

client/client-toolkits/pom.xml

@@ -4,7 +4,7 @@
     <parent>
         <groupId>com.aliyun.dataworks</groupId>
         <artifactId>client</artifactId>
-        <version>1.1.9-10</version>
+        <version>1.1.9-16</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

client/migrationx/migrationx-common/pom.xml

@@ -22,7 +22,7 @@
     <parent>
         <artifactId>migrationx</artifactId>
         <groupId>com.aliyun.dataworks</groupId>
-        <version>1.1.9-10</version>
+        <version>1.1.9-16</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

client/migrationx/migrationx-common/src/main/java/com/aliyun/migrationx/common/metrics/DefaultMetricCollector.java

@@ -2,8 +2,6 @@
 
 import java.util.function.Consumer;
 
-import com.aliyun.migrationx.common.metrics.DolphinMetricsCollector.Summary;
-
 /**
  * default implement
  * @author 聿剑
@@ -55,6 +53,16 @@ public void markTempSpecProcess(Metrics metrics) {
 
     }
 
+    @Override
+    public void markUnSupportedSpecProcess(Metrics metrics) {
+
+    }
+
+    @Override
+    public void incrementType(String taskType) {
+
+    }
+
     @Override
     public void markSuccessSpecProcess(String workflowName, String nodeName) {
 
@@ -81,7 +89,7 @@ public void finishCollector() {
     }
 
     @Override
-    public void finishCollector(Consumer<Summary> c) {
+    public void finishCollector(Consumer<Object> c) {
 
     }
 

client/migrationx/migrationx-common/src/main/java/com/aliyun/migrationx/common/metrics/DolphinMetrics.java

@@ -15,7 +15,6 @@
 
 package com.aliyun.migrationx.common.metrics;
 
-import lombok.Builder;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.ToString;

client/migrationx/migrationx-common/src/main/java/com/aliyun/migrationx/common/metrics/DolphinMetricsCollector.java

@@ -49,6 +49,8 @@ public class DolphinMetricsCollector implements MetricsCollector {
     private final Set<Metrics> failedDolphinSchedulerTasks = new HashSet<>();
     private final Set<Metrics> skippedDolphinSchedulerTasks = new HashSet<>();
     private final Set<Metrics> tempDolphinSchedulerTasks = new HashSet<>();
+    private final Set<Metrics> unSupportedDolphinSchedulerTasks = new HashSet<>();
+    private final Map<String, Integer> typeCounter = new HashMap<>();
 
     private String transformerType;
     private Date startTime;
@@ -158,6 +160,21 @@ public void markTempSpecProcess(Metrics tmp) {
         tempDolphinSchedulerTasks.add(metrics);
     }
 
+    @Override
+    public void markUnSupportedSpecProcess(Metrics metrics) {
+        unSupportedDolphinSchedulerTasks.add(metrics);
+    }
+
+    @Override
+    public void incrementType(String taskType) {
+        Integer count = typeCounter.get(taskType);
+        if (count == null) {
+            typeCounter.put(taskType, 1);
+        } else {
+            typeCounter.put(taskType, count + 1);
+        }
+    }
+
     private Metrics findMetrics(Metrics tmp) {
         String workflowName = tmp.getWorkflowName();
         String nodeName = tmp.getDwName();
@@ -176,21 +193,27 @@ public void finishCollector() {
     }
 
     @Override
-    public void finishCollector(Consumer<Summary> c) {
+    public void finishCollector(Consumer<Object> c) {
         Summary summary = new Summary();
         summary.setStartTime(startTime);
         summary.setEndTime(new Date());
         summary.setTransformerType(transformerType);
         summary.setType(CollectorType.DolphinScheduler.name());
-        summary.setSuccess(toSummary(successDolphinSchedulerTasks));
-        summary.setFailed(toSummary(failedDolphinSchedulerTasks));
-        summary.setSkipped(toSummary(skippedDolphinSchedulerTasks));
-        summary.setTemp(toSummary(tempDolphinSchedulerTasks));
         summary.setTotalSuccess(successDolphinSchedulerTasks.size());
         summary.setTotalFailed(failedDolphinSchedulerTasks.size());
         summary.setTotalSkipped(skippedDolphinSchedulerTasks.size());
         summary.setTotalTemp(tempDolphinSchedulerTasks.size());
+        summary.setTotalUnSupported(unSupportedDolphinSchedulerTasks.size());
         c.accept(summary);
+
+        Detail detail = new Detail();
+        detail.setSuccess(toSummary(successDolphinSchedulerTasks));
+        detail.setFailed(toSummary(failedDolphinSchedulerTasks));
+        detail.setSkipped(toSummary(skippedDolphinSchedulerTasks));
+        detail.setTemp(toSummary(tempDolphinSchedulerTasks));
+        detail.setUnSupported(toSummary(unSupportedDolphinSchedulerTasks));
+        detail.setTypeCounter(typeCounter);
+        c.accept(detail);
     }
 
     private List<Project> toSummary(Set<Metrics> metricsList) {
@@ -219,6 +242,7 @@ private List<Project> toSummary(Set<Metrics> metricsList) {
                 task = new Task();
                 task.setTaskCode(dolphinMetrics.getTaskCode());
                 task.setTaskName(dolphinMetrics.getTaskName());
+                task.setTaskType(dolphinMetrics.getTaskType());
                 taskMap.put(dolphinMetrics.getTaskCode(), task);
                 process.getTasks().add(task);
             }
@@ -249,10 +273,17 @@ public static class Summary {
         private int totalFailed;
         private int totalSkipped;
         private int totalTemp;
+        private int totalUnSupported;
+    }
+
+    @Data
+    public static class Detail {
         private List<Project> success;
         private List<Project> failed;
         private List<Project> skipped;
         private List<Project> temp;
+        private List<Project> unSupported;
+        private Map<String, Integer> typeCounter;
     }
 
     @Data
@@ -273,5 +304,6 @@ public static class Process {
     public static class Task {
         private String taskName;
         private Long taskCode;
+        private String taskType;
     }
 }

client/migrationx/migrationx-common/src/main/java/com/aliyun/migrationx/common/metrics/Metrics.java

@@ -15,6 +15,8 @@
 
 package com.aliyun.migrationx.common.metrics;
 
+import java.util.Objects;
+
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
@@ -34,4 +36,14 @@ public class Metrics {
     protected Long projectId;
     protected String region;
     protected String errorMsg;
+
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) {return false;}
+        Metrics metrics = (Metrics) o;
+        return Objects.equals(dwType, metrics.dwType) && Objects.equals(dwName, metrics.dwName) && Objects.equals(workflowName, metrics.workflowName);
+    }
+
+    public int hashCode() {
+        return Objects.hash(dwType, dwName, workflowName);
+    }
 }

client/migrationx/migrationx-common/src/main/java/com/aliyun/migrationx/common/metrics/MetricsCollector.java

@@ -17,8 +17,6 @@
 
 import java.util.function.Consumer;
 
-import com.aliyun.migrationx.common.metrics.DolphinMetricsCollector.Summary;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -45,6 +43,10 @@ public interface MetricsCollector {
 
     void markTempSpecProcess(Metrics metrics);
 
+    void markUnSupportedSpecProcess(Metrics metrics);
+
+    void incrementType(String taskType);
+
     void markSuccessSpecProcess(String workflowName, String nodeName);
 
     void markSuccessSpecProcess(Metrics metrics);
@@ -55,7 +57,7 @@ public interface MetricsCollector {
 
     void finishCollector();
 
-    void finishCollector(Consumer<Summary> c);
+    void finishCollector(Consumer<Object> c);
 
     Progress progress();
 }