[Feature] Add a userId field to ApplicationLog to record the operator (…

…#3221)

Co-authored-by: gongzhongqiang <[email protected]>

streampark-console/streampark-console-service/src/main/assembly/script/upgrade/mysql/2.2.2.sql

@@ -24,3 +24,6 @@ alter table `t_flink_app`
     add column `k8s_name` varchar(63) collate utf8mb4_general_ci default null,
     -- modify_time change with duration #3188
     modify column `modify_time` datetime not null default current_timestamp comment 'modify time';
+
+alter table `t_flink_log`
+    add column `user_id` bigint default null comment 'operator user id';

streampark-console/streampark-console-service/src/main/assembly/script/upgrade/pgsql/2.2.2.sql

@@ -17,3 +17,8 @@
 
 alter table "public"."t_flink_app"
     add column "k8s_name" varchar(63) collate "pg_catalog"."default";
+
+alter table "public"."t_flink_log"
+    add column "user_id" int8 collate "pg_catalog"."default";
+
+comment on column "public"."t_flink_log"."user_id" is 'operator user id';

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationBackUp.java

@@ -47,8 +47,6 @@ public class ApplicationBackUp {
 
   private transient boolean backup;
 
-  public ApplicationBackUp() {}
-
   public ApplicationBackUp(Application application) {
     this.appId = application.getId();
     this.sqlId = application.getSqlId();

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/ApplicationLog.java

@@ -44,7 +44,8 @@ public class ApplicationLog {
   private Integer optionName;
   /** option time */
   private Date optionTime;
-
   /** exception at the start */
   private String exception;
+  /** The user who operates the application */
+  private Long userId;
 }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Effective.java

@@ -46,13 +46,4 @@ public class Effective {
   private Date createTime;
 
   private transient EffectiveTypeEnum effectiveType;
-
-  public Effective() {}
-
-  public Effective(Long appId, EffectiveTypeEnum type, Long targetId) {
-    this.appId = appId;
-    this.targetType = type.getType();
-    this.targetId = targetId;
-    this.createTime = new Date();
-  }
 }

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/entity/Message.java

@@ -48,8 +48,6 @@ public class Message {
 
   private Date createTime;
 
-  public Message() {}
-
   public Message(
       Long userId, Long appId, String title, String context, NoticeTypeEnum noticeTypeEnum) {
     this.userId = userId;

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/application/impl/ApplicationActionServiceImpl.java

@@ -238,6 +238,7 @@ public void cancel(Application appParam) throws Exception {
     applicationLog.setJobManagerUrl(application.getJobManagerUrl());
     applicationLog.setOptionTime(new Date());
     applicationLog.setYarnAppId(application.getClusterId());
+    applicationLog.setUserId(commonService.getUserId());
 
     if (appParam.getSavePointed()) {
       FlinkAppHttpWatcher.addSavepoint(application.getId());
@@ -407,6 +408,7 @@ public void start(Application appParam, boolean auto) throws Exception {
     applicationLog.setOptionName(OperationEnum.START.getValue());
     applicationLog.setAppId(application.getId());
     applicationLog.setOptionTime(new Date());
+    applicationLog.setUserId(commonService.getUserId());
 
     // set the latest to Effective, (it will only become the current effective at this time)
     applicationManageService.toEffective(application);

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java

@@ -184,6 +184,7 @@ public boolean buildApplication(Long appId, boolean forceBuild) {
     applicationLog.setOptionName(RELEASE.getValue());
     applicationLog.setAppId(app.getId());
     applicationLog.setOptionTime(new Date());
+    applicationLog.setUserId(commonService.getUserId());
 
     // check if you need to go through the build process (if the jar and pom have changed,
     // you need to go through the build process, if other common parameters are modified,

streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java

@@ -103,6 +103,8 @@ public class SavePointServiceImpl extends ServiceImpl<SavePointMapper, SavePoint
 
   @Autowired private FlinkAppHttpWatcher flinkAppHttpWatcher;
 
+  @Autowired private CommonServiceImpl commonService;
+
   private final ExecutorService executorService =
       new ThreadPoolExecutor(
           Runtime.getRuntime().availableProcessors() * 5,
@@ -175,6 +177,7 @@ public void trigger(Long appId, @Nullable String savepointPath, @Nullable Boolea
     applicationLog.setJobManagerUrl(application.getJobManagerUrl());
     applicationLog.setOptionTime(new Date());
     applicationLog.setYarnAppId(application.getClusterId());
+    applicationLog.setUserId(commonService.getUserId());
 
     FlinkAppHttpWatcher.addSavepoint(application.getId());
 

streampark-console/streampark-console-service/src/main/resources/db/schema-h2.sql

@@ -155,6 +155,7 @@ create table if not exists `t_flink_log` (
   `exception` text ,
   `option_time` datetime default null,
   `option_name` tinyint default null,
+  `user_id` bigint default null comment 'operator user id',
   primary key(`id`)
 );