[DSIP-79][Task] Add datavines task for data quality

.github/workflows/e2e.yml

@@ -137,6 +137,8 @@ jobs:
             class: org.apache.dolphinscheduler.e2e.cases.HiveDataSourceE2ETest
           - name: DolphinDBDataSourceE2ETest
             class: org.apache.dolphinscheduler.e2e.cases.DolphinDBDataSourceE2ETest
+          - name: DatavinesTaskE2ETest
+            class: org.apache.dolphinscheduler.e2e.cases.tasks.DatavinesTaskE2ETest
     env:
       RECORDING_PATH: /tmp/recording-${{ matrix.case.name }}
     steps:

.github/workflows/lychee.toml

@@ -76,7 +76,7 @@ include_mail = false
 include_verbatim = false
 
 # Enable the checking of fragments in links.
-include_fragments = false
+include_fragments = "none"
 
 # User agent to send with each request.
 user_agent = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Safari/537.36"

docs/docs/en/guide/task/datavines.md

@@ -0,0 +1,30 @@
+# Datavines
+
+## Overview
+
+Use `Datavines Task` to create a datavines-type task and support data quality job in Datavines. When the worker executes `Datavines Task`,
+it will call `Datavines API` to trigger datavines job. Click [here](https://datavane.github.io/datavines-website/) for details about `Datavines`.
+
+## Create Task
+
+- Click Project Management-Project Name-Workflow Definition, and click the "Create Workflow" button to enter the DAG editing page.
+- Drag <img src="../../../../img/tasks/icons/datavines.png" width="15"/> from the toolbar to the canvas.
+
+## Task Parameter
+
+- Please refer to [DolphinScheduler Task Parameters Appendix](appendix.md) `Default Task Parameters` section for default parameters.
+
+|   **Parameter**   |                                            **Description**                                            |
+|-------------------|-------------------------------------------------------------------------------------------------------|
+| Datavines Address | The URL for the Datavines service, e.g., http://localhost:5600.                                       |
+| Datavines Job ID  | The unique job id for a datavines job.                                                                |
+| Datavines token   | The Datavines service access token can be obtained through token management on the Datavines service. |
+| Block on Failure  | When turned on, if the data quality check result is failed, the task result will be set as failed.    |
+
+## Task Example
+
+This example illustrates how to create a datavines task node.
+
+![demo-datavines](../../../../img/tasks/demo/datavines_task.png)
+
+![demo-get-datavines-job-id](../../../../img/tasks/demo/datavines_job_id.png)

docs/docs/zh/guide/task/datavines.md

@@ -0,0 +1,31 @@
+# Datavines
+
+## 综述
+
+`Datavines`任务类型，用于创建并执行 `Datavines` 类型任务来执行 Datavines 中的数据质量检查作业。Worker 执行该任务的时候，会通过 `Datavines API` 触发 `Datavines 的作业`。
+点击 [这里](https://datavane.github.io/datavines-website/) 获取更多关于 `Datavines` 的信息。
+
+## 创建任务
+
+- 点击项目管理-项目名称-工作流定义，点击"创建工作流"按钮，进入DAG编辑页面。
+- 工具栏中拖动 <img src="../../../../img/tasks/icons/datavines.png" width="15"/> 到画板中，即可完成创建。
+
+## 任务参数
+
+- 默认参数说明请参考[DolphinScheduler任务参数附录](appendix.md)`默认任务参数`一栏。
+
+|    **任务参数**     |                        **描述**                        |
+|-----------------|------------------------------------------------------|
+| Datavines 地址    | Datavines 服务的 url，例如：`http://localhost:5600`。        |
+| Datavines 作业 ID | Datavines 作业对应的唯一ID。                                 |
+| Datavines token | Datavines 服务访问 token, 可在 Datavines 服务上的 token 管理中取得。 |
+| 检查失败时阻塞         | 开启时，数据质量检查结果为失败时会将任务结果置为失败。                          |
+
+## 例子
+
+这个示例展示了如何创建 Datavines 任务节点：
+
+![demo-datavines](../../../../img/tasks/demo/datavines_task.png)
+
+![demo-get-datavines-job-id](../../../../img/tasks/demo/datavines_job_id.png)
+

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/configuration/TaskTypeConfiguration.java

@@ -46,6 +46,7 @@ public class TaskTypeConfiguration {
     private List<String> cloud;
     private List<String> logic;
     private List<String> dataIntegration;
+    private List<String> dataQuality;
     private List<String> other;
     private List<String> machineLearning;
 
@@ -61,6 +62,7 @@ public List<FavTaskDto> getDefaultTaskTypes() {
         logic.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_LOGIC)));
         dataIntegration
                 .forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_DATA_INTEGRATION)));
+        dataQuality.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_DATA_QUALITY)));
         machineLearning
                 .forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_MACHINE_LEARNING)));
         other.forEach(task -> defaultTaskTypes.add(new FavTaskDto(task, false, Constants.TYPE_OTHER)));
@@ -72,6 +74,7 @@ public void printDefaultTypes() {
         log.info("support default cloud task types: {}", cloud);
         log.info("support default logic task types: {}", logic);
         log.info("support default dataIntegration task types: {}", dataIntegration);
+        log.info("support default dataQuality task types: {}", dataQuality);
         log.info("support default machineLearning task types: {}", machineLearning);
         log.info("support default other task types: {}", other);
     }

dolphinscheduler-api/src/main/resources/task-type-config.yaml

@@ -47,6 +47,8 @@ task:
     - 'SEATUNNEL'
     - 'DATAX'
     - 'SQOOP'
+  dataQuality:
+    - 'DATAVINES'
   machineLearning:
     - 'JUPYTER'
     - 'MLFLOW'

dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/constants/Constants.java

@@ -306,6 +306,7 @@ public final class Constants {
     public static final String TYPE_LOGIC = "Logic";
     public static final String TYPE_OTHER = "Other";
     public static final String TYPE_MACHINE_LEARNING = "MachineLearning";
+    public static final String TYPE_DATA_QUALITY = "DataQuality";
 
     /**
      * spi constants

dolphinscheduler-e2e/dolphinscheduler-e2e-case/src/test/java/org/apache/dolphinscheduler/e2e/cases/tasks/DatavinesTaskE2ETest.java

@@ -0,0 +1,184 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.dolphinscheduler.e2e.cases.tasks;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.apache.dolphinscheduler.e2e.cases.workflow.BaseWorkflowE2ETest;
+import org.apache.dolphinscheduler.e2e.core.DolphinScheduler;
+import org.apache.dolphinscheduler.e2e.core.WebDriverHolder;
+import org.apache.dolphinscheduler.e2e.pages.LoginPage;
+import org.apache.dolphinscheduler.e2e.pages.project.ProjectPage;
+import org.apache.dolphinscheduler.e2e.pages.project.workflow.TaskInstanceTab;
+import org.apache.dolphinscheduler.e2e.pages.project.workflow.WorkflowDefinitionTab;
+import org.apache.dolphinscheduler.e2e.pages.project.workflow.WorkflowForm;
+import org.apache.dolphinscheduler.e2e.pages.project.workflow.WorkflowInstanceTab;
+import org.apache.dolphinscheduler.e2e.pages.project.workflow.task.DatavinesTaskForm;
+import org.apache.dolphinscheduler.e2e.pages.security.SecurityPage;
+import org.apache.dolphinscheduler.e2e.pages.security.TenantPage;
+import org.apache.dolphinscheduler.e2e.pages.security.UserPage;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.MethodOrderer;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestMethodOrder;
+import org.junitpioneer.jupiter.DisableIfTestFails;
+
+@TestMethodOrder(MethodOrderer.OrderAnnotation.class)
+@DolphinScheduler(composeFiles = "docker/datavines-task/docker-compose.yaml")
+@DisableIfTestFails
+public class DatavinesTaskE2ETest extends BaseWorkflowE2ETest {
+
+    private static final String mockServerAddress = "http://mockServer:1080";
+
+    private static final String jobId = "1";
+
+    private static final String token = "test-token";
+
+    @BeforeAll
+    public static void setup() {
+        browser = WebDriverHolder.getWebDriver();
+
+        TenantPage tenantPage = new LoginPage(browser)
+                .login(adminUser)
+                .goToNav(SecurityPage.class)
+                .goToTab(TenantPage.class);
+
+        if (tenantPage.tenants().stream().noneMatch(tenant -> tenant.tenantCode().equals(adminUser.getTenant()))) {
+            tenantPage
+                    .create(adminUser.getTenant())
+                    .goToNav(SecurityPage.class)
+                    .goToTab(UserPage.class)
+                    .update(adminUser);
+        }
+
+        tenantPage
+                .goToNav(ProjectPage.class)
+                .createProjectUntilSuccess(projectName);
+    }
+
+    @Test
+    @Order(10)
+    void testRunDatavinesTasks_SuccessCase() {
+        WorkflowDefinitionTab workflowDefinitionPage =
+                new ProjectPage(browser)
+                        .goToNav(ProjectPage.class)
+                        .goTo(projectName)
+                        .goToTab(WorkflowDefinitionTab.class);
+
+        String workflowName = "DatavinesSuccessCase";
+        String taskName = "DatavinesSuccessTask";
+        workflowDefinitionPage
+                .createWorkflow()
+                .<DatavinesTaskForm>addTask(WorkflowForm.TaskType.DATAVINES)
+                .address(mockServerAddress)
+                .jobId(jobId)
+                .token(token)
+                .name(taskName)
+                .submit()
+
+                .submit()
+                .name(workflowName)
+                .submit();
+
+        untilWorkflowDefinitionExist(workflowName);
+
+        workflowDefinitionPage.publish(workflowName);
+
+        runWorkflow(workflowName);
+        untilWorkflowInstanceExist(workflowName);
+        WorkflowInstanceTab.Row workflowInstance = untilWorkflowInstanceSuccess(workflowName);
+        assertThat(workflowInstance.executionTime()).isEqualTo(1);
+
+        TaskInstanceTab.Row taskInstance = untilTaskInstanceSuccess(workflowName, taskName);
+        assertThat(taskInstance.retryTimes()).isEqualTo(0);
+    }
+
+    @Test
+    @Order(20)
+    void testRunDatavinesTasks_FailureBlockDisabledCase() {
+        WorkflowDefinitionTab workflowDefinitionPage =
+                new ProjectPage(browser)
+                        .goToNav(ProjectPage.class)
+                        .goTo(projectName)
+                        .goToTab(WorkflowDefinitionTab.class);
+
+        String workflowName = "DatavinesFailureBlockDisabledCase";
+        String taskName = "DatavinesFailureBlockDisabledTask";
+        workflowDefinitionPage
+                .createWorkflow()
+                .<DatavinesTaskForm>addTask(WorkflowForm.TaskType.DATAVINES)
+                .address(mockServerAddress)
+                .jobId("3")
+                .token(token)
+                .name(taskName)
+                .submit()
+
+                .submit()
+                .name(workflowName)
+                .submit();
+
+        untilWorkflowDefinitionExist(workflowName);
+
+        workflowDefinitionPage.publish(workflowName);
+
+        runWorkflow(workflowName);
+        untilWorkflowInstanceExist(workflowName);
+        WorkflowInstanceTab.Row workflowInstance = untilWorkflowInstanceSuccess(workflowName);
+        assertThat(workflowInstance.executionTime()).isEqualTo(1);
+
+        TaskInstanceTab.Row taskInstance = untilTaskInstanceSuccess(workflowName, taskName);
+        assertThat(taskInstance.retryTimes()).isEqualTo(0);
+    }
+
+    @Test
+    @Order(30)
+    void testRunDatavinesTasks_FailureCase() {
+        WorkflowDefinitionTab workflowDefinitionPage =
+                new ProjectPage(browser)
+                        .goToNav(ProjectPage.class)
+                        .goTo(projectName)
+                        .goToTab(WorkflowDefinitionTab.class);
+
+        String workflowName = "DatavinesFailureCase";
+        String taskName = "DatavinesFailureTask";
+        // Job ID "2" is configured in MockServer to return execution status FAILURE
+        workflowDefinitionPage
+                .createWorkflow()
+                .<DatavinesTaskForm>addTask(WorkflowForm.TaskType.DATAVINES)
+                .address(mockServerAddress)
+                .jobId("2")
+                .token(token)
+                .name(taskName)
+                .submit()
+
+                .submit()
+                .name(workflowName)
+                .submit();
+
+        untilWorkflowDefinitionExist(workflowName);
+
+        workflowDefinitionPage.publish(workflowName);
+
+        runWorkflow(workflowName);
+        untilWorkflowInstanceExist(workflowName);
+        WorkflowInstanceTab.Row workflowInstance = untilWorkflowInstanceFailed(workflowName);
+        assertThat(workflowInstance.executionTime()).isEqualTo(1);
+    }
+}

dolphinscheduler-e2e/dolphinscheduler-e2e-case/src/test/java/org/apache/dolphinscheduler/e2e/pages/project/workflow/WorkflowForm.java

@@ -18,6 +18,7 @@
 package org.apache.dolphinscheduler.e2e.pages.project.workflow;
 
 import org.apache.dolphinscheduler.e2e.core.WebDriverWaitFactory;
+import org.apache.dolphinscheduler.e2e.pages.project.workflow.task.DatavinesTaskForm;
 import org.apache.dolphinscheduler.e2e.pages.project.workflow.task.HttpTaskForm;
 import org.apache.dolphinscheduler.e2e.pages.project.workflow.task.JavaTaskForm;
 import org.apache.dolphinscheduler.e2e.pages.project.workflow.task.PythonTaskForm;
@@ -67,6 +68,11 @@ public WorkflowForm(WebDriver driver) {
     @SneakyThrows
     @SuppressWarnings("unchecked")
     public <T> T addTask(TaskType type) {
+        if (type == TaskType.DATAVINES) {
+            driver.findElement(By.cssSelector(".task-cate-dq .n-collapse-item__header")).click();
+            WebDriverWaitFactory.createWebDriverWait(driver)
+                    .until(ExpectedConditions.visibilityOfElementLocated(By.className("task-item-DATAVINES")));
+        }
         final WebElement task = driver.findElement(By.className("task-item-" + type.name()));
         final WebElement canvas = driver.findElement(By.className("dag-container"));
 
@@ -90,6 +96,8 @@ public <T> T addTask(TaskType type) {
                 return (T) new JavaTaskForm(this);
             case PYTHON:
                 return (T) new PythonTaskForm(this);
+            case DATAVINES:
+                return (T) new DatavinesTaskForm(this);
         }
         throw new UnsupportedOperationException("Unknown task type");
     }
@@ -129,6 +137,7 @@ public enum TaskType {
         SWITCH,
         HTTP,
         JAVA,
-        PYTHON
+        PYTHON,
+        DATAVINES
     }
 }