You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2023/03/23 05:39:28 UTC
[incubator-streampark] branch dev updated: [Improve] record more info of application log (#2472)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
The following commit(s) were added to refs/heads/dev by this push:
new cbd6fbfce [Improve] record more info of application log (#2472)
cbd6fbfce is described below
commit cbd6fbfceee11e9323dba0e7972d478b1a4728a0
Author: totoro <zh...@foxmail.com>
AuthorDate: Thu Mar 23 13:39:22 2023 +0800
[Improve] record more info of application log (#2472)
* [Improve] record more info of application log
* [fix] use enum for operation type
---------
Co-authored-by: zhoulii <zh...@hikvision.com.cn>
Co-authored-by: benjobs <be...@apache.org>
---
.../main/assembly/script/schema/mysql-schema.sql | 1 +
.../main/assembly/script/schema/pgsql-schema.sql | 3 +-
.../main/assembly/script/upgrade/mysql/2.1.0.sql | 2 +
.../main/assembly/script/upgrade/pgsql/2.1.0.sql | 3 ++
.../ApplicationBuildPipelineController.java | 16 +++++-
.../core/controller/ApplicationController.java | 7 +++
.../console/core/entity/ApplicationLog.java | 3 +-
.../Operation.java} | 26 +++++++---
.../console/core/service/AppBuildPipeService.java | 4 +-
.../core/service/ApplicationLogService.java | 2 +
.../core/service/impl/AppBuildPipeServiceImpl.java | 12 +++--
.../service/impl/ApplicationLogServiceImpl.java | 5 ++
.../core/service/impl/ApplicationServiceImpl.java | 25 +++++----
.../core/service/impl/SavePointServiceImpl.java | 27 ++++++----
.../src/main/resources/db/schema-h2.sql | 1 +
.../resources/mapper/core/ApplicationLogMapper.xml | 1 +
.../src/api/flink/app/app.ts | 6 +++
.../src/enums/flinkEnum.ts | 11 ++++
.../src/locales/lang/en/flink/app.ts | 1 +
.../src/locales/lang/zh-CN/flink/app.ts | 1 +
.../flink/app/components/AppDetail/DetailTab.vue | 59 ++++++++++++++++------
.../src/views/flink/app/data/detail.data.ts | 1 +
22 files changed, 166 insertions(+), 51 deletions(-)
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/schema/mysql-schema.sql b/streampark-console/streampark-console-service/src/main/assembly/script/schema/mysql-schema.sql
index 8c1ac27e5..a4d1d0475 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/schema/mysql-schema.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/schema/mysql-schema.sql
@@ -171,6 +171,7 @@ create table `t_flink_log` (
`success` tinyint default null,
`exception` text collate utf8mb4_general_ci,
`option_time` datetime default null,
+ `option_name` tinyint default null,
primary key (`id`) using btree
) engine=innodb auto_increment=100000 default charset=utf8mb4 collate=utf8mb4_general_ci;
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/schema/pgsql-schema.sql b/streampark-console/streampark-console-service/src/main/assembly/script/schema/pgsql-schema.sql
index c85a7939e..49e31df86 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/schema/pgsql-schema.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/schema/pgsql-schema.sql
@@ -407,7 +407,8 @@ create table "public"."t_flink_log" (
"job_manager_url" varchar(255) collate "pg_catalog"."default",
"success" int2,
"exception" text collate "pg_catalog"."default",
- "option_time" timestamp(6)
+ "option_time" timestamp(6),
+ "option_name" int2
)
;
alter table "public"."t_flink_log" add constraint "t_flink_log_pkey" primary key ("id");
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.1.0.sql b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.1.0.sql
index 2d3f59609..eb77a76b1 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.1.0.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.1.0.sql
@@ -90,4 +90,6 @@ create table `t_yarn_queue` (
-- Issue-2324 End --
+alter table `t_flink_log` add column `option_name` tinyint default null;
+
set foreign_key_checks = 1;
diff --git a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.1.0.sql b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.1.0.sql
index cfb197d8f..c2e864df7 100644
--- a/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.1.0.sql
+++ b/streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.1.0.sql
@@ -106,3 +106,6 @@ alter table "public"."t_yarn_queue" add constraint "t_yarn_queue_pkey" primary k
alter table "public"."t_yarn_queue" add constraint "unique_team_id_queue_label" unique("team_id", "queue_label");
-- Issue-2324 End --
+
+
+alter table "public"."t_flink_log" add column "option_name" type int2;
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationBuildPipelineController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationBuildPipelineController.java
index debaf46c6..ca9136885 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationBuildPipelineController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationBuildPipelineController.java
@@ -24,7 +24,10 @@ import org.apache.streampark.console.core.annotation.ApiAccess;
import org.apache.streampark.console.core.bean.AppBuildDockerResolvedDetail;
import org.apache.streampark.console.core.entity.AppBuildPipeline;
import org.apache.streampark.console.core.entity.Application;
+import org.apache.streampark.console.core.entity.ApplicationLog;
+import org.apache.streampark.console.core.enums.Operation;
import org.apache.streampark.console.core.service.AppBuildPipeService;
+import org.apache.streampark.console.core.service.ApplicationLogService;
import org.apache.streampark.console.core.service.ApplicationService;
import org.apache.streampark.console.core.service.FlinkSqlService;
import org.apache.streampark.flink.packer.pipeline.DockerResolvedSnapshot;
@@ -43,6 +46,7 @@ import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
+import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
@@ -60,6 +64,8 @@ public class ApplicationBuildPipelineController {
@Autowired private FlinkSqlService flinkSqlService;
+ @Autowired private ApplicationLogService applicationLogService;
+
/**
* Release application building pipeline.
*
@@ -93,6 +99,12 @@ public class ApplicationBuildPipelineController {
public RestResponse buildApplication(Long appId, boolean forceBuild) {
try {
Application app = applicationService.getById(appId);
+
+ ApplicationLog applicationLog = new ApplicationLog();
+ applicationLog.setOptionName(Operation.RELEASE.getValue());
+ applicationLog.setAppId(app.getId());
+ applicationLog.setOptionTime(new Date());
+
boolean envOk = applicationService.checkEnv(app);
if (!envOk) {
throw new ApiAlertException(
@@ -108,6 +120,8 @@ public class ApplicationBuildPipelineController {
// you don't need to go through the build process)
boolean needBuild = applicationService.checkBuildAndUpdate(app);
if (!needBuild) {
+ applicationLog.setSuccess(true);
+ applicationLogService.save(applicationLog);
return RestResponse.success(true);
}
@@ -116,7 +130,7 @@ public class ApplicationBuildPipelineController {
flinkSqlService.rollback(app);
}
- boolean actionResult = appBuildPipeService.buildApplication(app);
+ boolean actionResult = appBuildPipeService.buildApplication(app, applicationLog);
return RestResponse.success(actionResult);
} catch (Exception e) {
return RestResponse.success(false).message(e.getMessage());
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java
index 2607fad29..f999c9347 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java
@@ -348,6 +348,13 @@ public class ApplicationController {
return RestResponse.success(applicationList);
}
+ @PostMapping("deleteOperationLog")
+ @RequiresPermissions("app:delete")
+ public RestResponse deleteOperationLog(ApplicationLog applicationLog) {
+ Boolean deleted = applicationLogService.delete(applicationLog);
+ return RestResponse.success(deleted);
+ }
+
@PostMapping("delete")
@RequiresPermissions("app:delete")
public RestResponse delete(Application app) throws InternalException {
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationLog.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationLog.java
index 5fab750f4..ade3195f8 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationLog.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationLog.java
@@ -40,7 +40,8 @@ public class ApplicationLog {
private String jobManagerUrl;
/** start status */
private Boolean success;
-
+ /** option name */
+ private Integer optionName;
/** option time */
private Date optionTime;
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationLogService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/enums/Operation.java
similarity index 62%
copy from streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationLogService.java
copy to streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/enums/Operation.java
index 05ab7a471..2acc60c08 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationLogService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/enums/Operation.java
@@ -15,17 +15,27 @@
* limitations under the License.
*/
-package org.apache.streampark.console.core.service;
+package org.apache.streampark.console.core.enums;
-import org.apache.streampark.console.base.domain.RestRequest;
-import org.apache.streampark.console.core.entity.ApplicationLog;
+import lombok.Getter;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.service.IService;
+import java.io.Serializable;
+import java.util.Arrays;
-public interface ApplicationLogService extends IService<ApplicationLog> {
+@Getter
+public enum Operation implements Serializable {
+ RELEASE(0),
+ START(1),
+ SAVEPOINT(2),
+ CANCEL(3);
- IPage<ApplicationLog> page(ApplicationLog applicationLog, RestRequest request);
+ private final int value;
- void removeApp(Long appId);
+ Operation(int value) {
+ this.value = value;
+ }
+
+ public static Operation of(Integer option) {
+ return Arrays.stream(values()).filter((x) -> x.value == option).findFirst().orElse(null);
+ }
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/AppBuildPipeService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/AppBuildPipeService.java
index 2f8342071..48e92c60a 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/AppBuildPipeService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/AppBuildPipeService.java
@@ -19,6 +19,7 @@ package org.apache.streampark.console.core.service;
import org.apache.streampark.console.core.entity.AppBuildPipeline;
import org.apache.streampark.console.core.entity.Application;
+import org.apache.streampark.console.core.entity.ApplicationLog;
import org.apache.streampark.flink.packer.pipeline.DockerResolvedSnapshot;
import org.apache.streampark.flink.packer.pipeline.PipelineStatus;
@@ -33,7 +34,8 @@ import java.util.Optional;
public interface AppBuildPipeService extends IService<AppBuildPipeline> {
/** Build application. This is an async call method. */
- boolean buildApplication(@Nonnull Application app) throws Exception;
+ boolean buildApplication(@Nonnull Application app, ApplicationLog applicationLog)
+ throws Exception;
/**
* Get current build pipeline instance of specified application
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationLogService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationLogService.java
index 05ab7a471..62585e71d 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationLogService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationLogService.java
@@ -28,4 +28,6 @@ public interface ApplicationLogService extends IService<ApplicationLog> {
IPage<ApplicationLog> page(ApplicationLog applicationLog, RestRequest request);
void removeApp(Long appId);
+
+ Boolean delete(ApplicationLog applicationLog);
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
index e24f94c76..5c092dfd4 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
@@ -30,6 +30,7 @@ import org.apache.streampark.console.base.util.WebUtils;
import org.apache.streampark.console.core.entity.AppBuildPipeline;
import org.apache.streampark.console.core.entity.Application;
import org.apache.streampark.console.core.entity.ApplicationConfig;
+import org.apache.streampark.console.core.entity.ApplicationLog;
import org.apache.streampark.console.core.entity.FlinkEnv;
import org.apache.streampark.console.core.entity.FlinkSql;
import org.apache.streampark.console.core.entity.Message;
@@ -41,6 +42,7 @@ import org.apache.streampark.console.core.mapper.ApplicationBuildPipelineMapper;
import org.apache.streampark.console.core.service.AppBuildPipeService;
import org.apache.streampark.console.core.service.ApplicationBackUpService;
import org.apache.streampark.console.core.service.ApplicationConfigService;
+import org.apache.streampark.console.core.service.ApplicationLogService;
import org.apache.streampark.console.core.service.ApplicationService;
import org.apache.streampark.console.core.service.CommonService;
import org.apache.streampark.console.core.service.FlinkEnvService;
@@ -115,6 +117,8 @@ public class AppBuildPipeServiceImpl
@Autowired private ApplicationService applicationService;
+ @Autowired private ApplicationLogService applicationLogService;
+
@Autowired private FlinkRESTAPIWatcher flinkRESTAPIWatcher;
@Autowired private ApplicationConfigService applicationConfigService;
@@ -139,8 +143,7 @@ public class AppBuildPipeServiceImpl
Caffeine.newBuilder().expireAfterWrite(30, TimeUnit.DAYS).build();
@Override
- public boolean buildApplication(@Nonnull Application app) {
-
+ public boolean buildApplication(@Nonnull Application app, ApplicationLog applicationLog) {
// 1) flink sql setDependency
FlinkSql newFlinkSql = flinkSqlService.getCandidate(app.getId(), CandidateType.NEW);
FlinkSql effectiveFlinkSql = flinkSqlService.getEffective(app.getId(), false);
@@ -262,7 +265,7 @@ public class AppBuildPipeServiceImpl
backUpService.backup(app, null);
}
}
-
+ applicationLog.setSuccess(true);
app.setBuild(false);
} else {
@@ -277,8 +280,11 @@ public class AppBuildPipeServiceImpl
app.setRelease(ReleaseState.FAILED.get());
app.setOptionState(OptionState.NONE.getValue());
app.setBuild(true);
+ applicationLog.setException(Utils.stringifyException(snapshot.error().exception()));
+ applicationLog.setSuccess(false);
}
applicationService.updateRelease(app);
+ applicationLogService.save(applicationLog);
if (flinkRESTAPIWatcher.isWatchingApp(app.getId())) {
flinkRESTAPIWatcher.init();
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationLogServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationLogServiceImpl.java
index c4ae64ce0..7155020b4 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationLogServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationLogServiceImpl.java
@@ -55,4 +55,9 @@ public class ApplicationLogServiceImpl extends ServiceImpl<ApplicationLogMapper,
new LambdaQueryWrapper<ApplicationLog>().eq(ApplicationLog::getAppId, appId);
this.remove(queryWrapper);
}
+
+ @Override
+ public Boolean delete(ApplicationLog applicationLog) {
+ return removeById(applicationLog.getId());
+ }
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
index 386b5f4a3..e35760813 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
@@ -55,6 +55,7 @@ import org.apache.streampark.console.core.enums.ChangedType;
import org.apache.streampark.console.core.enums.CheckPointType;
import org.apache.streampark.console.core.enums.ConfigFileType;
import org.apache.streampark.console.core.enums.FlinkAppState;
+import org.apache.streampark.console.core.enums.Operation;
import org.apache.streampark.console.core.enums.OptionState;
import org.apache.streampark.console.core.enums.ReleaseState;
import org.apache.streampark.console.core.mapper.ApplicationMapper;
@@ -1168,8 +1169,15 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
public void cancel(Application appParam) throws Exception {
FlinkRESTAPIWatcher.setOptionState(appParam.getId(), OptionState.CANCELLING);
Application application = getById(appParam.getId());
-
application.setState(FlinkAppState.CANCELLING.getValue());
+
+ ApplicationLog applicationLog = new ApplicationLog();
+ applicationLog.setOptionName(Operation.CANCEL.getValue());
+ applicationLog.setAppId(application.getId());
+ applicationLog.setJobManagerUrl(application.getJobManagerUrl());
+ applicationLog.setOptionTime(new Date());
+ applicationLog.setYarnAppId(application.getClusterId());
+
if (appParam.getSavePointed()) {
FlinkRESTAPIWatcher.addSavepoint(application.getId());
application.setOptionState(OptionState.SAVEPOINTING.getValue());
@@ -1251,6 +1259,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
10L,
TimeUnit.MINUTES,
cancelResponse -> {
+ applicationLog.setSuccess(true);
if (cancelResponse != null && cancelResponse.savePointDir() != null) {
String savePointDir = cancelResponse.savePointDir();
log.info("savePoint path: {}", savePointDir);
@@ -1289,14 +1298,9 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
FlinkRESTAPIWatcher.unWatching(application.getId());
}
- ApplicationLog log = new ApplicationLog();
- log.setAppId(application.getId());
- log.setYarnAppId(application.getClusterId());
- log.setOptionTime(new Date());
String exception = Utils.stringifyException(e);
- log.setException(exception);
- log.setSuccess(false);
- applicationLogService.save(log);
+ applicationLog.setException(exception);
+ applicationLog.setSuccess(false);
}
})
.whenComplete(
@@ -1306,6 +1310,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
application.getK8sNamespace(), application.getJobName());
}
cancelFutureMap.remove(application.getId());
+ applicationLogService.save(applicationLog);
});
}
@@ -1394,6 +1399,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
String flinkUserJar = null;
String jobId = new JobID().toHexString();
ApplicationLog applicationLog = new ApplicationLog();
+ applicationLog.setOptionName(Operation.START.getValue());
applicationLog.setAppId(application.getId());
applicationLog.setOptionTime(new Date());
@@ -1588,7 +1594,6 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
}
applicationLog.setSuccess(true);
- applicationLogService.save(applicationLog);
// set savepoint to expire
savePointService.expire(application.getId());
},
@@ -1599,7 +1604,6 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
String exception = Utils.stringifyException(e);
applicationLog.setException(exception);
applicationLog.setSuccess(false);
- applicationLogService.save(applicationLog);
Application app = getById(appParam.getId());
app.setState(FlinkAppState.FAILED.getValue());
app.setOptionState(OptionState.NONE.getValue());
@@ -1613,6 +1617,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
})
.whenComplete(
(t, e) -> {
+ applicationLogService.save(applicationLog);
startFutureMap.remove(application.getId());
});
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java
index f36eeb2a1..93a6567b5 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java
@@ -36,6 +36,7 @@ import org.apache.streampark.console.core.entity.FlinkCluster;
import org.apache.streampark.console.core.entity.FlinkEnv;
import org.apache.streampark.console.core.entity.SavePoint;
import org.apache.streampark.console.core.enums.CheckPointType;
+import org.apache.streampark.console.core.enums.Operation;
import org.apache.streampark.console.core.enums.OptionState;
import org.apache.streampark.console.core.mapper.SavePointMapper;
import org.apache.streampark.console.core.service.ApplicationConfigService;
@@ -281,6 +282,13 @@ public class SavePointServiceImpl extends ServiceImpl<SavePointMapper, SavePoint
log.info("Start to trigger savepoint for app {}", appId);
Application application = applicationService.getById(appId);
+ ApplicationLog applicationLog = new ApplicationLog();
+ applicationLog.setOptionName(Operation.SAVEPOINT.getValue());
+ applicationLog.setAppId(application.getId());
+ applicationLog.setJobManagerUrl(application.getJobManagerUrl());
+ applicationLog.setOptionTime(new Date());
+ applicationLog.setYarnAppId(application.getClusterId());
+
FlinkRESTAPIWatcher.addSavepoint(application.getId());
application.setOptionState(OptionState.SAVEPOINTING.getValue());
@@ -311,38 +319,35 @@ public class SavePointServiceImpl extends ServiceImpl<SavePointMapper, SavePoint
CompletableFuture<SavepointResponse> savepointFuture =
CompletableFuture.supplyAsync(() -> FlinkClient.triggerSavepoint(request), executorService);
- handleSavepointResponseFuture(application, savepointFuture);
+ handleSavepointResponseFuture(application, applicationLog, savepointFuture);
}
private void handleSavepointResponseFuture(
- Application application, CompletableFuture<SavepointResponse> savepointFuture) {
+ Application application,
+ ApplicationLog applicationLog,
+ CompletableFuture<SavepointResponse> savepointFuture) {
CompletableFutureUtils.runTimeout(
savepointFuture,
10L,
TimeUnit.MINUTES,
savepointResponse -> {
if (savepointResponse != null && savepointResponse.savePointDir() != null) {
+ applicationLog.setSuccess(true);
String savePointDir = savepointResponse.savePointDir();
log.info("Request savepoint successful, savepointDir: {}", savePointDir);
}
},
e -> {
log.error("Trigger savepoint for flink job failed.", e);
- ApplicationLog log = new ApplicationLog();
- log.setAppId(application.getId());
- if (ExecutionMode.isYarnMode(application.getExecutionMode())) {
- log.setYarnAppId(application.getClusterId());
- }
- log.setOptionTime(new Date());
String exception = Utils.stringifyException(e);
- log.setException(exception);
+ applicationLog.setException(exception);
if (!(e instanceof TimeoutException)) {
- log.setSuccess(false);
+ applicationLog.setSuccess(false);
}
- applicationLogService.save(log);
})
.whenComplete(
(t, e) -> {
+ applicationLogService.save(applicationLog);
application.setOptionState(OptionState.NONE.getValue());
application.setOptionTime(new Date());
applicationService.update(application);
diff --git a/streampark-console/streampark-console-service/src/main/resources/db/schema-h2.sql b/streampark-console/streampark-console-service/src/main/resources/db/schema-h2.sql
index 8c9a5f5fc..896f0181c 100644
--- a/streampark-console/streampark-console-service/src/main/resources/db/schema-h2.sql
+++ b/streampark-console/streampark-console-service/src/main/resources/db/schema-h2.sql
@@ -152,6 +152,7 @@ create table if not exists `t_flink_log` (
`success` tinyint default null,
`exception` text ,
`option_time` datetime default null,
+ `option_name` tinyint default null,
primary key(`id`)
);
diff --git a/streampark-console/streampark-console-service/src/main/resources/mapper/core/ApplicationLogMapper.xml b/streampark-console/streampark-console-service/src/main/resources/mapper/core/ApplicationLogMapper.xml
index 20bef89e3..fb0eb52da 100644
--- a/streampark-console/streampark-console-service/src/main/resources/mapper/core/ApplicationLogMapper.xml
+++ b/streampark-console/streampark-console-service/src/main/resources/mapper/core/ApplicationLogMapper.xml
@@ -26,6 +26,7 @@
<result column="success" jdbcType="INTEGER" property="success"/>
<result column="exception" jdbcType="LONGVARCHAR" property="exception"/>
<result column="option_time" jdbcType="DATE" property="optionTime"/>
+ <result column="option_name" jdbcType="INTEGER" property="optionName"/>
</resultMap>
</mapper>
diff --git a/streampark-console/streampark-console-webapp/src/api/flink/app/app.ts b/streampark-console/streampark-console-webapp/src/api/flink/app/app.ts
index 7a9cc7b2d..93f49226e 100644
--- a/streampark-console/streampark-console-webapp/src/api/flink/app/app.ts
+++ b/streampark-console/streampark-console-webapp/src/api/flink/app/app.ts
@@ -46,6 +46,7 @@ enum APP_API {
ROLLBACK = '/flink/app/rollback',
REVOKE = '/flink/app/revoke',
OPTION_LOG = '/flink/app/optionlog',
+ DELETE_OPERATION_LOG = '/flink/app/deleteOperationLog',
CHECK_JAR = '/flink/app/checkjar',
VERIFY_SCHEMA = '/flink/app/verifySchema',
CHECK_SAVEPOINT_PATH = '/flink/app/checkSavepointPath',
@@ -159,6 +160,11 @@ export function fetchBackUps(data) {
export function fetchOptionLog(data) {
return defHttp.post({ url: APP_API.OPTION_LOG, data });
}
+
+export function fetchDeleteOperationLog(id: string) {
+ return defHttp.post({ url: APP_API.DELETE_OPERATION_LOG, data: { id } });
+}
+
/**
* forced stop
* @param params id:string
diff --git a/streampark-console/streampark-console-webapp/src/enums/flinkEnum.ts b/streampark-console/streampark-console-webapp/src/enums/flinkEnum.ts
index 2bc19252b..b664e601c 100644
--- a/streampark-console/streampark-console-webapp/src/enums/flinkEnum.ts
+++ b/streampark-console/streampark-console-webapp/src/enums/flinkEnum.ts
@@ -62,6 +62,17 @@ export enum ReleaseStateEnum {
REVOKED = 10,
}
+export enum OperationEnum {
+ /** user operation: release app */
+ RELEASE = 0,
+ /** user operation: start app */
+ START = 1,
+ /** user operation: trigger savepoint */
+ SAVEPOINT = 2,
+ /** user operation: cancel app */
+ CANCEL = 3,
+}
+
export enum OptionStateEnum {
/** Application which is currently action: none. */
NONE = 0,
diff --git a/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/app.ts b/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/app.ts
index 2cb9ebae2..d07a69397 100644
--- a/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/app.ts
+++ b/streampark-console/streampark-console-webapp/src/locales/lang/en/flink/app.ts
@@ -101,6 +101,7 @@ export default {
confDeleteTitle: 'Are you sure delete this record?',
sqlDeleteTitle: 'Are you sure delete this sql?',
confBackupTitle: 'Are you sure delete this backup?',
+ operationLogDeleteTitle: 'Are you sure delete this operation log?',
copyPath: 'Copy Path',
pointDeleteTitle: 'Are you sure delete?',
copySuccess: 'copied to clipboard successfully',
diff --git a/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/app.ts b/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/app.ts
index d66f0a4f3..7e6b53b30 100644
--- a/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/app.ts
+++ b/streampark-console/streampark-console-webapp/src/locales/lang/zh-CN/flink/app.ts
@@ -99,6 +99,7 @@ export default {
confDeleteTitle: '您确定要删除此记录吗?',
sqlDeleteTitle: '您确定要删除此 SQL 吗?',
confBackupTitle: '您确定要删除该备份吗?',
+ operationLogDeleteTitle: '您确定要删除该操作记录吗?',
copyPath: '复制路径',
pointDeleteTitle: '您确定要删除?',
copySuccess: '已成功复制到剪贴板',
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppDetail/DetailTab.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppDetail/DetailTab.vue
index aee63e13e..72d3fc037 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppDetail/DetailTab.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppDetail/DetailTab.vue
@@ -42,7 +42,12 @@
import { fetchGetVer, fetchListVer, fetchRemoveConf } from '/@/api/flink/config';
import { fetchRemoveSavePoint, fetchSavePonitHistory } from '/@/api/flink/app/savepoint';
- import { fetchBackUps, fetchOptionLog, fetchRemoveBackup } from '/@/api/flink/app/app';
+ import {
+ fetchBackUps,
+ fetchOptionLog,
+ fetchRemoveBackup,
+ fetchDeleteOperationLog,
+ } from '/@/api/flink/app/app';
import { decodeByBase64 } from '/@/utils/cipher';
import { useModal } from '/@/components/Modal';
import CompareModal from './CompareModal.vue';
@@ -136,7 +141,7 @@
...tableCommonConf,
});
- const [registerLogsTable] = useTable({
+ const [registerLogsTable, { reload: reloadOperationLog }] = useTable({
api: fetchOptionLog,
columns: getOptionLogColumns(),
...tableCommonConf,
@@ -300,6 +305,31 @@
color: 'error'
}]
}
+
+ function getOperationLogAction(record: Recordable): ActionItem[] {
+ return [
+ {
+ tooltip: { title: t('flink.app.detail.detailTab.exception') },
+ auth: 'app:detail',
+ shape: 'circle',
+ type: 'default',
+ ifShow: !record.success,
+ icon: 'ant-design:eye-outlined',
+ onClick: handleException.bind(null, record),
+ },
+ {
+ popConfirm: {
+ title: t('flink.app.detail.detailTab.operationLogDeleteTitle'),
+ confirm: handleDeleteOperationLog.bind(null, record),
+ },
+ auth: 'app:delete',
+ type: 'link',
+ icon: 'ant-design:delete-outlined',
+ color: 'error',
+ },
+ ];
+ }
+
/* delete savePoint */
async function handleDeleteSavePoint(record: Recordable) {
await fetchRemoveSavePoint({ id: record.id });
@@ -311,6 +341,11 @@
reloadBackup();
}
+ async function handleDeleteOperationLog(record: Recordable) {
+ await fetchDeleteOperationLog(record.id);
+ reloadOperationLog();
+ }
+
/* copy path */
function handleCopy(record: Recordable) {
try {
@@ -437,6 +472,12 @@
<TabPane key="6" :tab="t('flink.app.detail.detailTab.detailTabName.operationLog')" v-if="tabConf.showOptionLog">
<BasicTable @register="registerLogsTable">
<template #bodyCell="{ column, record }">
+ <template v-if="column.dataIndex == 'optionName'">
+ <Tag color="#52c41a" v-if="record.optionName === OperationEnum.RELEASE"> Release </Tag>
+ <Tag color="#2db7f5" v-if="record.optionName === OperationEnum.START"> Start </Tag>
+ <Tag color="#108ee9" v-if="record.optionName === OperationEnum.SAVEPOINT"> Savepoint </Tag>
+ <Tag color="#0C7EF2" v-if="record.optionName === OperationEnum.CANCEL"> Cancel </Tag>
+ </template>
<template v-if="column.dataIndex == 'yarnAppId'">
<a-button type="link" @click="handleView(app as any, '')">
{{ record.yarnAppId }}
@@ -456,19 +497,7 @@
<Tag class="start-state" color="#f5222d" v-else> FAILED </Tag>
</template>
<template v-if="column.dataIndex == 'operation'">
- <TableAction
- :actions="[
- {
- tooltip: { title: t('flink.app.detail.detailTab.exception') },
- auth: 'app:detail',
- shape: 'circle',
- type: 'default',
- ifShow: !record.success,
- icon: 'ant-design:eye-outlined',
- onClick: handleException.bind(null, record),
- },
- ]"
- />
+ <TableAction :actions="getOperationLogAction(record)" />
</template>
</template>
</BasicTable>
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/data/detail.data.ts b/streampark-console/streampark-console-webapp/src/views/flink/app/data/detail.data.ts
index c70678fe8..17af2a939 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/data/detail.data.ts
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/data/detail.data.ts
@@ -131,6 +131,7 @@ export const getBackupColumns = (): BasicColumn[] => [
];
export const getOptionLogColumns = (): BasicColumn[] => [
+ { title: 'Operation Name', dataIndex: 'optionName' },
{ title: 'Application Id', dataIndex: 'yarnAppId' },
{ title: 'JobManager URL', dataIndex: 'jobManagerUrl' },
{ title: 'Start Status', dataIndex: 'success', width: 120 },