Merge remote-tracking branch 'origin/dev' into dev

dlink-admin/src/main/java/com/dlink/service/impl/TaskServiceImpl.java

@@ -30,6 +30,7 @@ import com.dlink.result.SqlExplainResult;
 import com.dlink.service.*;
 import com.dlink.utils.CustomStringJavaCompiler;
 import com.dlink.utils.JSONUtil;
+
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
@@ -89,7 +90,7 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
         Asserts.checkNull(task, Tips.TASK_NOT_EXIST);
         if (Dialect.isSql(task.getDialect())) {
             return executeCommonSql(SqlDTO.build(task.getStatement(),
-                    task.getDatabaseId(), null));
+                task.getDatabaseId(), null));
         }
         JobConfig config = buildJobConfig(task);
         JobManager jobManager = JobManager.build(config);
@@ -107,7 +108,7 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
         task.setStep(JobLifeCycle.ONLINE.getValue());
         if (Dialect.isSql(task.getDialect())) {
             return executeCommonSql(SqlDTO.build(task.getStatement(),
-                    task.getDatabaseId(), null));
+                task.getDatabaseId(), null));
         }
         JobConfig config = buildJobConfig(task);
         JobManager jobManager = JobManager.build(config);
@@ -127,7 +128,7 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
         }
         if (Dialect.isSql(task.getDialect())) {
             return executeCommonSql(SqlDTO.build(task.getStatement(),
-                    task.getDatabaseId(), null));
+                task.getDatabaseId(), null));
         }
         task.setSavePointStrategy(SavePointStrategy.LATEST.getValue());
         JobConfig config = buildJobConfig(task);
@@ -236,7 +237,7 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
     public boolean saveOrUpdateTask(Task task) {
         // to compiler java udf
         if (Asserts.isNotNullString(task.getDialect()) && Dialect.JAVA.equalsVal(task.getDialect())
-                && Asserts.isNotNullString(task.getStatement())) {
+            && Asserts.isNotNullString(task.getStatement())) {
             CustomStringJavaCompiler compiler = new CustomStringJavaCompiler(task.getStatement());
             task.setSavePointPath(compiler.getFullClassName());
         }
@@ -245,8 +246,8 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
             Task taskInfo = getById(task.getId());
             Assert.check(taskInfo);
             if (JobLifeCycle.RELEASE.equalsValue(taskInfo.getStep()) ||
-                    JobLifeCycle.ONLINE.equalsValue(taskInfo.getStep()) ||
-                    JobLifeCycle.CANCEL.equalsValue(taskInfo.getStep())) {
+                JobLifeCycle.ONLINE.equalsValue(taskInfo.getStep()) ||
+                JobLifeCycle.CANCEL.equalsValue(taskInfo.getStep())) {
                 throw new BusException("该作业已" + JobLifeCycle.get(taskInfo.getStep()).getLabel() + "，禁止修改！");
             }
             task.setStep(JobLifeCycle.DEVELOP.getValue());
@@ -526,15 +527,18 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
                 Savepoints latestSavepoints = savepointsService.getLatestSavepointByTaskId(task.getId());
                 if (Asserts.isNotNull(latestSavepoints)) {
                     config.setSavePointPath(latestSavepoints.getPath());
+                    config.getConfig().put("execution.savepoint.path", latestSavepoints.getPath());
                 }
                 break;
             case EARLIEST:
                 Savepoints earliestSavepoints = savepointsService.getEarliestSavepointByTaskId(task.getId());
                 if (Asserts.isNotNull(earliestSavepoints)) {
                     config.setSavePointPath(earliestSavepoints.getPath());
+                    config.getConfig().put("execution.savepoint.path", earliestSavepoints.getPath());
                 }
                 break;
             case CUSTOM:
+                config.getConfig().put("execution.savepoint.path", config.getSavePointPath());
                 break;
             default:
                 config.setSavePointPath(null);
@@ -572,7 +576,7 @@ public class TaskServiceImpl extends SuperServiceImpl<TaskMapper, Task> implemen
 
     private boolean inRefreshPlan(JobInstance jobInstance) {
         if ((!JobStatus.isDone(jobInstance.getStatus())) || (Asserts.isNotNull(jobInstance.getFinishTime())
-                && Duration.between(jobInstance.getFinishTime(), LocalDateTime.now()).toMinutes() < 1)) {
+            && Duration.between(jobInstance.getFinishTime(), LocalDateTime.now()).toMinutes() < 1)) {
             return true;
         } else {
             return false;

dlink-executor/src/main/java/com/dlink/executor/AppBatchExecutor.java

 package com.dlink.executor;
 
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
+import com.dlink.assertion.Asserts;
+
 /**
  * AppBatchExecutor
  *
@@ -12,7 +16,12 @@ public class AppBatchExecutor extends Executor {
 
     public AppBatchExecutor(ExecutorSetting executorSetting) {
         this.executorSetting = executorSetting;
-        this.environment = StreamExecutionEnvironment.createLocalEnvironment();
+        if (Asserts.isNotNull(executorSetting.getConfig())) {
+            Configuration configuration = Configuration.fromMap(executorSetting.getConfig());
+            this.environment = StreamExecutionEnvironment.createLocalEnvironment(configuration);
+        } else {
+            this.environment = StreamExecutionEnvironment.createLocalEnvironment();
+        }
         init();
     }
 

dlink-executor/src/main/java/com/dlink/executor/AppStreamExecutor.java

 package com.dlink.executor;
 
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
+import com.dlink.assertion.Asserts;
+
 /**
  * AppStreamExecutor
  *
@@ -12,7 +16,12 @@ public class AppStreamExecutor extends Executor {
 
     public AppStreamExecutor(ExecutorSetting executorSetting) {
         this.executorSetting = executorSetting;
-        this.environment = StreamExecutionEnvironment.getExecutionEnvironment();
+        if (Asserts.isNotNull(executorSetting.getConfig())) {
+            Configuration configuration = Configuration.fromMap(executorSetting.getConfig());
+            this.environment = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
+        } else {
+            this.environment = StreamExecutionEnvironment.getExecutionEnvironment();
+        }
         init();
     }
 

dlink-executor/src/main/java/com/dlink/executor/LocalBatchExecutor.java

 package com.dlink.executor;
 
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
+import com.dlink.assertion.Asserts;
+
 /**
  * LocalBatchExecutor
  *
@@ -12,7 +16,12 @@ public class LocalBatchExecutor extends Executor {
 
     public LocalBatchExecutor(ExecutorSetting executorSetting) {
         this.executorSetting = executorSetting;
-        this.environment = StreamExecutionEnvironment.createLocalEnvironment();
+        if (Asserts.isNotNull(executorSetting.getConfig())) {
+            Configuration configuration = Configuration.fromMap(executorSetting.getConfig());
+            this.environment = StreamExecutionEnvironment.createLocalEnvironment(configuration);
+        } else {
+            this.environment = StreamExecutionEnvironment.createLocalEnvironment();
+        }
         init();
     }
 

dlink-executor/src/main/java/com/dlink/executor/LocalStreamExecutor.java

 package com.dlink.executor;
 
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
+import com.dlink.assertion.Asserts;
+
 /**
  * LocalStreamExecuter
  *
@@ -12,7 +16,12 @@ public class LocalStreamExecutor extends Executor {
 
     public LocalStreamExecutor(ExecutorSetting executorSetting) {
         this.executorSetting = executorSetting;
-        this.environment = StreamExecutionEnvironment.createLocalEnvironment();
+        if (Asserts.isNotNull(executorSetting.getConfig())) {
+            Configuration configuration = Configuration.fromMap(executorSetting.getConfig());
+            this.environment = StreamExecutionEnvironment.createLocalEnvironment(configuration);
+        } else {
+            this.environment = StreamExecutionEnvironment.createLocalEnvironment();
+        }
         init();
     }
 

dlink-executor/src/main/java/com/dlink/executor/RemoteBatchExecutor.java

 package com.dlink.executor;
 
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
+import com.dlink.assertion.Asserts;
+
 /**
  * RemoteBatchExecutor
  *
@@ -13,7 +17,12 @@ public class RemoteBatchExecutor extends Executor {
     public RemoteBatchExecutor(EnvironmentSetting environmentSetting, ExecutorSetting executorSetting) {
         this.environmentSetting = environmentSetting;
         this.executorSetting = executorSetting;
-        this.environment = StreamExecutionEnvironment.createRemoteEnvironment(environmentSetting.getHost(), environmentSetting.getPort());
+        if (Asserts.isNotNull(executorSetting.getConfig())) {
+            Configuration configuration = Configuration.fromMap(executorSetting.getConfig());
+            this.environment = StreamExecutionEnvironment.createRemoteEnvironment(environmentSetting.getHost(), environmentSetting.getPort(), configuration);
+        } else {
+            this.environment = StreamExecutionEnvironment.createRemoteEnvironment(environmentSetting.getHost(), environmentSetting.getPort());
+        }
         init();
     }
 

dlink-executor/src/main/java/com/dlink/executor/RemoteStreamExecutor.java

 package com.dlink.executor;
 
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.DeploymentOptions;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 
+import com.dlink.assertion.Asserts;
+
 /**
  * RemoteStreamExecutor
  *
@@ -13,7 +17,12 @@ public class RemoteStreamExecutor extends Executor {
     public RemoteStreamExecutor(EnvironmentSetting environmentSetting, ExecutorSetting executorSetting) {
         this.environmentSetting = environmentSetting;
         this.executorSetting = executorSetting;
-        this.environment = StreamExecutionEnvironment.createRemoteEnvironment(environmentSetting.getHost(), environmentSetting.getPort());
+        if (Asserts.isNotNull(executorSetting.getConfig())) {
+            Configuration configuration = Configuration.fromMap(executorSetting.getConfig());
+            this.environment = StreamExecutionEnvironment.createRemoteEnvironment(environmentSetting.getHost(), environmentSetting.getPort(), configuration);
+        } else {
+            this.environment = StreamExecutionEnvironment.createRemoteEnvironment(environmentSetting.getHost(), environmentSetting.getPort());
+        }
         init();
     }
 

docker-compose.yaml

+version: '3'
+# 默认使用 flink 1.13.3
+
+services:
+  # 配置 dinky 后端服务
+  dinky-server:
+    image: sivacohan/dinky-server:lastest
+    build:
+      context: .
+      dockerfile: ./docker/server/Dockerfile
+#    command: [ "/bin/sh", "-c", "java -Dloader.path=./lib,./plugins -Ddruid.mysql.usePingMethod=false -jar -Xms512M -Xmx2048M ./dlink-admin-*.jar" ]
+#    volumes:
+#      # 挂载dinky lib/plugins 目录
+#      - ./docker/dinky/lib:/app/lib
+#      - ./docker/dinky/plugins:/app/plugins
+    environment:
+      SPRING_DATASOURCE_URL: jdbc:mysql://database:3306/dlink?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&useSSL=false&zeroDateTimeBehavior=convertToNull&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true
+    depends_on:
+      - database
+    restart: unless-stopped
+
+  # 配置 dinky web 界面
+  dinky-web:
+    image: sivacohan/dinky-web:lastest
+    build:
+      context: .
+      dockerfile: ./docker/web/Dockerfile
+    environment:
+      # API_HOST dinky 后端服务地址 (可以是IP、或者Hostname)
+      # API_HOST dinky 后端服务端口
+      API_HOST: dinky-server
+      API_PORT: 8888
+    ports:
+      - "8080:80"
+    depends_on:
+      - dinky-server
+      - database
+
+  # 配置 mysql5.7 数据库
+  database:
+    image: sivacohan/dinky-mysql-server:5.7
+    build:
+      context: .
+      dockerfile: ./docker/mysql/Dockerfile
+    command: mysqld --character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ci
+    ports:
+      - 3306:3306
+    volumes:
+      - mysql-data:/var/lib/mysql
+    environment:
+      MYSQL_ROOT_PASSWORD: 123456
+      MYSQL_DATABASE: dlink
+      MYSQL_USER: dlink
+      MYSQL_PASSWORD: dlink
+    healthcheck:
+      test: [ "CMD", "mysqladmin" ,"ping", "-h", "localhost" ]
+      timeout: 20s
+      retries: 10
+    restart: unless-stopped
+
+
+volumes:
+  mysql-data:
\ No newline at end of file

docker/mysql/Dockerfile

+FROM mysql:5.7 as production-stage
+COPY ./dlink-doc/sql/dlink.sql /docker-entrypoint-initdb.d/01-dlink.sql
+COPY ./dlink-doc/sql/dlink_history.sql /docker-entrypoint-initdb.d/02-dlink_history.sql

docker/server/Dockerfile

+# stage 0, 依赖缓存
+FROM maven:3.8.4-jdk-8 as deps-stage
+WORKDIR /app
+ADD ./pom.xml ./pom.xml
+ADD ./dlink-core/pom.xml ./dlink-core/pom.xml
+ADD ./dlink-executor/pom.xml ./dlink-executor/pom.xml
+ADD ./dlink-connectors/pom.xml ./dlink-connectors/pom.xml
+ADD ./dlink-connectors/dlink-connector-jdbc-1.14/pom.xml ./dlink-connectors/dlink-connector-jdbc-1.14/pom.xml
+ADD ./dlink-connectors/dlink-connector-jdbc-1.13/pom.xml ./dlink-connectors/dlink-connector-jdbc-1.13/pom.xml
+ADD ./dlink-connectors/dlink-connector-jdbc-1.12/pom.xml ./dlink-connectors/dlink-connector-jdbc-1.12/pom.xml
+ADD ./dlink-connectors/dlink-connector-jdbc-1.11/pom.xml ./dlink-connectors/dlink-connector-jdbc-1.11/pom.xml
+ADD ./dlink-alert/pom.xml ./dlink-alert/pom.xml
+ADD ./dlink-alert/dlink-alert-base/pom.xml ./dlink-alert/dlink-alert-base/pom.xml
+ADD ./dlink-alert/dlink-alert-wechat/pom.xml ./dlink-alert/dlink-alert-wechat/pom.xml
+ADD ./dlink-alert/dlink-alert-dingtalk/pom.xml ./dlink-alert/dlink-alert-dingtalk/pom.xml
+ADD ./dlink-metadata/pom.xml ./dlink-metadata/pom.xml
+ADD ./dlink-metadata/dlink-metadata-postgresql/pom.xml ./dlink-metadata/dlink-metadata-postgresql/pom.xml
+ADD ./dlink-metadata/dlink-metadata-clickhouse/pom.xml ./dlink-metadata/dlink-metadata-clickhouse/pom.xml
+ADD ./dlink-metadata/dlink-metadata-mysql/pom.xml ./dlink-metadata/dlink-metadata-mysql/pom.xml
+ADD ./dlink-metadata/dlink-metadata-phoenix/pom.xml ./dlink-metadata/dlink-metadata-phoenix/pom.xml
+ADD ./dlink-metadata/dlink-metadata-sqlserver/pom.xml ./dlink-metadata/dlink-metadata-sqlserver/pom.xml
+ADD ./dlink-metadata/dlink-metadata-hive/pom.xml ./dlink-metadata/dlink-metadata-hive/pom.xml
+ADD ./dlink-metadata/dlink-metadata-base/pom.xml ./dlink-metadata/dlink-metadata-base/pom.xml
+ADD ./dlink-metadata/dlink-metadata-doris/pom.xml ./dlink-metadata/dlink-metadata-doris/pom.xml
+ADD ./dlink-metadata/dlink-metadata-oracle/pom.xml ./dlink-metadata/dlink-metadata-oracle/pom.xml
+ADD ./dlink-admin/pom.xml ./dlink-admin/pom.xml
+ADD ./dlink-extends/pom.xml ./dlink-extends/pom.xml
+ADD ./dlink-app/pom.xml ./dlink-app/pom.xml
+ADD ./dlink-app/dlink-app-1.13/pom.xml ./dlink-app/dlink-app-1.13/pom.xml
+ADD ./dlink-app/dlink-app-1.14/pom.xml ./dlink-app/dlink-app-1.14/pom.xml
+ADD ./dlink-app/dlink-app-1.12/pom.xml ./dlink-app/dlink-app-1.12/pom.xml
+ADD ./dlink-app/dlink-app-base/pom.xml ./dlink-app/dlink-app-base/pom.xml
+ADD ./dlink-app/dlink-app-1.11/pom.xml ./dlink-app/dlink-app-1.11/pom.xml
+ADD ./dlink-gateway/pom.xml ./dlink-gateway/pom.xml
+ADD ./dlink-common/pom.xml ./dlink-common/pom.xml
+ADD ./dlink-daemon/pom.xml ./dlink-daemon/pom.xml
+ADD ./dlink-assembly/pom.xml ./dlink-assembly/pom.xml
+ADD ./dlink-client/pom.xml ./dlink-client/pom.xml
+ADD ./dlink-client/dlink-client-1.11/pom.xml ./dlink-client/dlink-client-1.11/pom.xml
+ADD ./dlink-client/dlink-client-base/pom.xml ./dlink-client/dlink-client-base/pom.xml
+ADD ./dlink-client/dlink-client-hadoop/pom.xml ./dlink-client/dlink-client-hadoop/pom.xml
+ADD ./dlink-client/dlink-client-1.12/pom.xml ./dlink-client/dlink-client-1.12/pom.xml
+ADD ./dlink-client/dlink-client-1.14/pom.xml ./dlink-client/dlink-client-1.14/pom.xml
+ADD ./dlink-client/dlink-client-1.13/pom.xml ./dlink-client/dlink-client-1.13/pom.xml
+ADD ./dlink-function/pom.xml ./dlink-function/pom.xml
+ADD ./dlink-web/pom.xml ./dlink-web/pom.xml
+RUN mvn -T 1C -B dependency:go-offline --fail-never || true
+
+# stage 0, 引入plugins
+FROM maven:3.8.4-jdk-8 as plugin-stage
+WORKDIR /app
+ADD https://archive.apache.org/dist/flink/flink-1.13.3/flink-1.13.3-bin-scala_2.12.tgz /tmp
+RUN tar zxvf /tmp/flink-1.13.3-bin-scala_2.12.tgz -C /app --strip-components=1
+
+
+# stage 1, build stage
+FROM maven:3.8.4-jdk-8 as build-stage
+WORKDIR /app
+COPY --from=deps-stage /root/.m2/repository /root/.m2/repository
+COPY --from=deps-stage /app/ /app
+ADD .  /app
+RUN mvn clean package -Dmaven.test.skip=true
+RUN mkdir /tmp/dist
+RUN tar zxvf ./build/dlink-release-*.tar.gz -C /tmp/dist --strip-components=1
+
+
+# stage 2, production stage
+FROM openjdk:8-oracle as production-stage
+WORKDIR /app
+RUN mkdir plugins
+COPY --from=build-stage /tmp/dist /app
+COPY --from=plugin-stage /app/lib /app/plugins
+
+
+CMD [ "/bin/sh", "-c", "java -Dloader.path=./lib,./plugins -Ddruid.mysql.usePingMethod=false -jar -Xms512M -Xmx2048M ./dlink-admin-*.jar" ]

docker/web/Dockerfile

+# stage 0, build stage
+FROM sivacohan/dinky-server:lastest as build-stage
+
+FROM nginx:1.21.6 AS production-stage
+WORKDIR /app
+COPY --from=build-stage /app/html /app
+COPY ./docker/web/dinky.conf.template /etc/nginx/templates/default.conf.template
\ No newline at end of file

docker/web/dinky.conf.template

+server {
+    listen       80;
+    listen       [::]:80;
+    server_name  localhost;
+    root         /app;
+    gzip on;
+    gzip_min_length 1k;
+    gzip_comp_level 9;
+    gzip_types text/plain application/javascript application/x-javascript text/css application/xml text/javascript application/x-httpd-php image/jpeg image/gif image/png;
+    gzip_vary on;
+    gzip_disable "MSIE [1-6]\.";
+
+    location / {
+        index  index.html index.htm;
+        try_files $uri $uri/ /index.html;
+    }
+    error_page 404 /404.html;
+    location = /404.html {
+    }
+
+    error_page 500 502 503 504 /50x.html;
+    location = /50x.html {
+    }
+    
+    location ^~ /api {
+        proxy_pass http://${API_HOST}:${API_PORT};
+        proxy_set_header   X-Forwarded-Proto $scheme;
+        proxy_set_header   X-Real-IP         $remote_addr;
+    }
+}