[1.10.0]starrock monitor update

linkis-extensions/linkis-et-monitor/src/main/java/org/apache/linkis/monitor/scheduled/JobHistoryMonitor.java

@@ -27,10 +27,7 @@
 import org.apache.linkis.monitor.jobhistory.errorcode.JobHistoryErrorCodeAlertSender;
 import org.apache.linkis.monitor.jobhistory.index.JobIndexRule;
 import org.apache.linkis.monitor.jobhistory.index.JobIndexSender;
-import org.apache.linkis.monitor.jobhistory.jobtime.JobTimeExceedAlertSender;
-import org.apache.linkis.monitor.jobhistory.jobtime.JobTimeExceedRule;
-import org.apache.linkis.monitor.jobhistory.jobtime.StarrocksTimeExceedAlterSender;
-import org.apache.linkis.monitor.jobhistory.jobtime.StarrocksTimeExceedRule;
+import org.apache.linkis.monitor.jobhistory.jobtime.*;
 import org.apache.linkis.monitor.jobhistory.labels.JobHistoryLabelsAlertSender;
 import org.apache.linkis.monitor.jobhistory.labels.JobHistoryLabelsRule;
 import org.apache.linkis.monitor.jobhistory.runtime.CommonJobRunTimeRule;
@@ -100,7 +97,7 @@ public void jobHistoryFinishedScan() {
       logger.info("Get JobHistoryId from cache ID:" + id);
     }
     List<DataFetcher> fetchers =
-        JobMonitorUtils.generateFetchersfortime(startTime, endTime, id, "updated_time");
+        JobMonitorUtils.generateFetchersfortime(startTime, endTime, id, "finished_job");
     if (fetchers.isEmpty()) {
       logger.warn("generated 0 dataFetchers, plz check input");
       return;
@@ -178,7 +175,7 @@ public void jobHistoryFinishedScan() {
     JobIndexRule jobIndexRule = new JobIndexRule(new JobIndexSender());
     scannerIndex.addScanRule(jobIndexRule);
     List<DataFetcher> createFetcher =
-        JobMonitorUtils.generateFetchersfortime(startTime, endTime, id, "department");
+        JobMonitorUtils.generateFetchersfortime(startTime, endTime, id, "");
     JobMonitorUtils.run(scannerIndex, createFetcher, true);
   }
 
@@ -195,7 +192,7 @@ public void jobHistoryUnfinishedScan() {
     AnomalyScanner scanner = new DefaultScanner();
     boolean shouldStart = false;
     List<DataFetcher> fetchers =
-        JobMonitorUtils.generateFetchers(startTime, endTime, maxIntervalMs, id, "created_time");
+        JobMonitorUtils.generateFetchers(startTime, endTime, maxIntervalMs, id, "unfinished_job");
     if (fetchers.isEmpty()) {
       logger.warn("generated 0 dataFetchers, plz check input");
       return;
@@ -215,9 +212,52 @@ public void jobHistoryUnfinishedScan() {
               jobTimeAlerts.keySet(), new JobTimeExceedAlertSender(jobTimeAlerts));
       scanner.addScanRule(jobTimeExceedRule);
     }
+    JobMonitorUtils.run(scanner, fetchers, shouldStart);
+  }
+
+  /** * 每10分钟扫描一次,扫描两个小时之内的任务，告警要求：管理台配置告警相关参数 */
+  @Scheduled(cron = "${linkis.monitor.jdbc.timeout.alert.cron:0 0/10 0 * * ?}")
+  public void jdbcUnfinishedAlertScan() {
+    long id =
+        Optional.ofNullable(CacheUtils.cacheBuilder.getIfPresent("jdbcUnfinishedAlertScan"))
+            .orElse(MonitorConfig.JOB_HISTORY_TIME_EXCEED.getValue());
+    long intervalMs = 7200 * 1000;
+    long maxIntervalMs = Constants.ERRORCODE_MAX_INTERVALS_SECONDS() * 1000;
+    long endTime = System.currentTimeMillis();
+    long startTime = endTime - intervalMs;
+    AnomalyScanner scanner = new DefaultScanner();
+    List<DataFetcher> fetchers =
+        JobMonitorUtils.generateFetchers(startTime, endTime, maxIntervalMs, id, "");
+    if (fetchers.isEmpty()) {
+      logger.warn("jdbcUnfinishedScan generated 0 dataFetchers, plz check input");
+      return;
+    }
     StarrocksTimeExceedRule starrocksTimeExceedRule =
-        new StarrocksTimeExceedRule(new StarrocksTimeExceedAlterSender());
+        new StarrocksTimeExceedRule(new StarrocksTimeExceedAlertSender());
     scanner.addScanRule(starrocksTimeExceedRule);
-    JobMonitorUtils.run(scanner, fetchers, shouldStart);
+    JobMonitorUtils.run(scanner, fetchers, true);
+  }
+
+  /** * 每10分钟扫描一次,扫描两个小时之内的任务，满足要求触发kill kill要求：数据源配置kill参数 */
+  @Scheduled(cron = "${linkis.monitor.jdbc.timeout.kill.cron:0 0/10 0 * * ?}")
+  public void jdbcUnfinishedKillScan() {
+    long id =
+            Optional.ofNullable(CacheUtils.cacheBuilder.getIfPresent("jdbcUnfinishedKillScan"))
+                    .orElse(MonitorConfig.JOB_HISTORY_TIME_EXCEED.getValue());
+    long intervalMs = 7200 * 1000;
+    long maxIntervalMs = Constants.ERRORCODE_MAX_INTERVALS_SECONDS() * 1000;
+    long endTime = System.currentTimeMillis();
+    long startTime = endTime - intervalMs;
+    AnomalyScanner scanner = new DefaultScanner();
+    List<DataFetcher> fetchers =
+            JobMonitorUtils.generateFetchers(startTime, endTime, maxIntervalMs, id, "");
+    if (fetchers.isEmpty()) {
+      logger.warn("jdbcUnfinishedScan generated 0 dataFetchers, plz check input");
+      return;
+    }
+    StarrocksTimeKillRule starrocksTimeKillRule =
+            new StarrocksTimeKillRule(new StarrocksTimeKillAlertSender());
+    scanner.addScanRule(starrocksTimeKillRule);
+    JobMonitorUtils.run(scanner, fetchers, true);
   }
 }

linkis-extensions/linkis-et-monitor/src/main/java/org/apache/linkis/monitor/until/JobMonitorUtils.java

@@ -44,14 +44,14 @@ public static void run(AnomalyScanner scanner, List<DataFetcher> fetchers, Boole
   }
 
   public static List<DataFetcher> generateFetchers(
-      long startTime, long endTime, long maxIntervalMs, long id, String timeType) {
+      long startTime, long endTime, long maxIntervalMs, long id, String jobStatus) {
     List<DataFetcher> ret = new ArrayList<>();
     long pe = endTime;
     long ps;
     while (pe > startTime) {
       ps = Math.max(pe - maxIntervalMs, startTime);
       String[] fetcherArgs =
-          new String[] {String.valueOf(ps), String.valueOf(pe), String.valueOf(id), timeType};
+          new String[] {String.valueOf(ps), String.valueOf(pe), String.valueOf(id), jobStatus};
       ret.add(new JobHistoryDataFetcher(fetcherArgs, MapperFactory.getJobHistoryMapper()));
       logger.info(
           "Generated dataFetcher for startTime: " + new Date(ps) + ". EndTime: " + new Date(pe));
@@ -61,11 +61,11 @@ public static List<DataFetcher> generateFetchers(
   }
 
   public static List<DataFetcher> generateFetchersfortime(
-      long startTime, long endTime, long id, String timeType) {
+      long startTime, long endTime, long id, String jobStatus) {
     List<DataFetcher> fetchers = new ArrayList<>();
     String[] fetcherArgs =
         new String[] {
-          String.valueOf(startTime), String.valueOf(endTime), String.valueOf(id), timeType
+          String.valueOf(startTime), String.valueOf(endTime), String.valueOf(id), jobStatus
         };
     fetchers.add(new JobHistoryDataFetcher(fetcherArgs, MapperFactory.getJobHistoryMapper()));
     logger.info(

linkis-extensions/linkis-et-monitor/src/main/resources/mapper/common/JobHistoryMapper.xml

@@ -127,7 +127,7 @@
         <include refid="jobhistory_query"/>
         FROM linkis_ps_job_history_group_history job JOIN linkis_org_user org ON job.submit_user = org.user_name
         <where>
-            <if test="id != null">job.id > #{id}</if>
+            <if test="id != null">job.id >= #{id}</if>
             <if test="umUser != null">and job.submit_user = #{umUser}</if>
             <if test="engineType != null">and job.engine_type = #{engineType}</if>
             <if test="startDate != null">and job.created_time >= #{startDate} AND job.created_time <![CDATA[<=]]>#{endDate}

linkis-extensions/linkis-et-monitor/src/main/scala/org/apache/linkis/monitor/jobhistory/JobHistoryDataFetcher.scala

@@ -54,60 +54,46 @@ class JobHistoryDataFetcher(args: Array[Any], mapper: JobHistoryMapper)
         "Wrong input for JobHistoryDataFetcher. DataType: " + args.getClass.getCanonicalName
       )
     }
-    if (args != null && args.length == 2) {
-      val start = Utils.tryCatch(args(0).asInstanceOf[String].toLong) { t =>
-        {
-          logger.error("Failed to get data from DB: Illegal arguments.", t)
-          throw t
-        }
-      }
-      val end = Utils.tryCatch(args(1).asInstanceOf[String].toLong) { t =>
-        {
-          logger.error("Failed to get data from DB: Illegal arguments.", t)
-          throw t
-        }
-      }
-      mapper
-        .search(null, null, null, new Date(start), new Date(end), null)
-        .asInstanceOf[util.List[scala.Any]]
-    } else if (args != null && args.length == 4) {
-      val start = Utils.tryCatch(args(0).asInstanceOf[String].toLong) { t =>
-        {
-          logger.error("Failed to get data from DB: Illegal arguments.", t)
-          throw t
-        }
-      }
-      val end = Utils.tryCatch(args(1).asInstanceOf[String].toLong) { t =>
-        {
-          logger.error("Failed to get data from DB: Illegal arguments.", t)
-          throw t
-        }
-      }
-      val id = Utils.tryCatch(args(2).asInstanceOf[String].toLong) { t =>
-        {
-          logger.error("Failed to get data from DB: Illegal arguments.", t)
-          throw t
-        }
-      }
-      if (
-          StringUtils.isNotBlank(args(3).asInstanceOf[String]) && args(3)
-            .asInstanceOf[String]
-            .equals("updated_time")
-      ) {
-        val list = new util.ArrayList[String]()
-        Constants.DATA_FINISHED_JOB_STATUS_ARRAY.foreach(list.add)
-        mapper
-          .searchByCacheAndUpdateTime(id, null, list, new Date(start), new Date(end), null)
-          .asInstanceOf[util.List[scala.Any]]
-      } else {
-        var list = new util.ArrayList[String]()
-        Constants.DATA_UNFINISHED_JOB_STATUS_ARRAY.foreach(list.add)
-        if (args(3).asInstanceOf[String].equals("department")) {
-          list = null;
-        }
-        mapper
-          .searchByCache(id, null, list, new Date(start), new Date(end), null)
-          .asInstanceOf[util.List[scala.Any]]
+    if (args != null) {
+      val start = args(0).asInstanceOf[String].toLong
+      val end = args(1).asInstanceOf[String].toLong
+      // 根据参数数量进行不同的处理
+      args.length match {
+        // 参数数量为2，则数据库查询仅筛选开始和结束时间
+        case 2 =>
+          mapper
+            .search(null, null, null, new Date(start), new Date(end), null)
+            .asInstanceOf[util.List[scala.Any]]
+        // 参数数量为4，根据第四个参数进行不同的查询
+        case 4 =>
+          val id = args(2).asInstanceOf[String].toLong
+          val parm = args(3).asInstanceOf[String]
+          parm match {
+            // 筛选任务包含id，时间，已完成状态任务
+            case "finished_job" =>
+              val list = new util.ArrayList[String]()
+              Constants.DATA_FINISHED_JOB_STATUS_ARRAY.foreach(list.add)
+              mapper
+                .searchByCacheAndUpdateTime(id, null, list, new Date(start), new Date(end), null)
+                .asInstanceOf[util.List[scala.Any]]
+            // 筛选任务包含id，时间，未完成状态任务
+            case "unfinished_job" =>
+              var list = new util.ArrayList[String]()
+              Constants.DATA_UNFINISHED_JOB_STATUS_ARRAY.foreach(list.add)
+              mapper
+                .searchByCache(id, null, list, new Date(start), new Date(end), null)
+                .asInstanceOf[util.List[scala.Any]]
+            // 筛选任务包含id，时间
+            case _ =>
+              mapper
+                .searchByCache(id, null, null, new Date(start), new Date(end), null)
+                .asInstanceOf[util.List[scala.Any]]
+          }
+        case _ =>
+          throw new AnomalyScannerException(
+            21304,
+            "Wrong input for JobHistoryDataFetcher. Data: " + args
+          )
       }
     } else {
       throw new AnomalyScannerException(

linkis-extensions/linkis-et-monitor/src/main/scala/org/apache/linkis/monitor/jobhistory/jobtime/JobTimeExceedRule.scala

@@ -73,7 +73,6 @@ class JobTimeExceedRule(thresholds: util.Set[String], hitObserver: Observer)
     val alertData: util.List[JobHistory] = new util.ArrayList[JobHistory]()
     for (sd <- data.asScala) {
       if (sd != null && sd.getData() != null) {
-        var idLong = 0L
         for (d <- sd.getData().asScala) {
           if (d.isInstanceOf[JobHistory]) {
             val jobHistory = d.asInstanceOf[JobHistory]
@@ -84,24 +83,11 @@ class JobTimeExceedRule(thresholds: util.Set[String], hitObserver: Observer)
                 alertData.add(d.asInstanceOf[JobHistory])
               }
             }
-            if (idLong == 0L || jobHistory.getId < idLong) {
-              idLong = jobHistory.getId
-            }
+            scanRuleList.put("jobhistoryScan", jobHistory.getId)
           } else {
             logger.warn("Ignored wrong input data Type : " + d + ", " + d.getClass.getCanonicalName)
           }
         }
-        if (idLong > 0L) {
-          val id = Optional
-            .ofNullable(CacheUtils.cacheBuilder.getIfPresent("jobhistoryScan"))
-            .orElse(MonitorConfig.JOB_HISTORY_TIME_EXCEED.getValue)
-          if (id == 0) {
-            scanRuleList.put("jobhistoryScan", idLong)
-          }
-          if (id > idLong) {
-            scanRuleList.put("jobhistoryScan", idLong)
-          }
-        }
       } else {
         logger.warn("Ignored null scanned data")
       }

linkis-extensions/linkis-et-monitor/src/main/scala/org/apache/linkis/monitor/jobhistory/jobtime/StarrocksTimeExceedAlterSender.scala → linkis-extensions/linkis-et-monitor/src/main/scala/org/apache/linkis/monitor/jobhistory/jobtime/StarrocksTimeExceedAlertSender.scala

@@ -32,7 +32,7 @@ import java.util
 
 import scala.collection.JavaConverters.asScalaBufferConverter
 
-class StarrocksTimeExceedAlterSender extends Observer with Logging {
+class StarrocksTimeExceedAlertSender extends Observer with Logging {
 
   /**
    * Observer Pattern

linkis-extensions/linkis-et-monitor/src/main/scala/org/apache/linkis/monitor/jobhistory/jobtime/StarrocksTimeExceedRule.scala

@@ -37,6 +37,8 @@ class StarrocksTimeExceedRule(hitObserver: Observer)
     extends AbstractScanRule(event = new StarrocksTimeExceedHitEvent, observer = hitObserver)
     with Logging {
 
+  private val scanRuleList = CacheUtils.cacheBuilder
+
   /**
    * if data match the pattern, return true and trigger observer should call isMatched()
    *
@@ -52,6 +54,7 @@ class StarrocksTimeExceedRule(hitObserver: Observer)
     val alertData: util.List[JobHistory] = new util.ArrayList[JobHistory]()
     for (scannedData <- data.asScala) {
       if (scannedData != null && scannedData.getData() != null) {
+        var taskMinID = 0L;
         for (jobHistory <- scannedData.getData().asScala) {
           jobHistory match {
             case job: JobHistory =>
@@ -80,27 +83,10 @@ class StarrocksTimeExceedRule(hitObserver: Observer)
                     alertData.add(job)
                   }
                 }
-                // 获取超时kill配置信息
-                if (StringUtils.isNotBlank(job.getParams)) {
-                  val connectParamsMap = MapUtils.getMap(
-                    datasourceConfMap,
-                    "connectParams",
-                    new util.HashMap[AnyRef, AnyRef]
-                  )
-                  val killTime = MapUtils.getString(connectParamsMap, "kill_task_time", "")
-                  logger.info("starock  killTime: {}", killTime)
-                  if (StringUtils.isNotBlank(killTime) && elapse > killTime.toLong * 60 * 1000) {
-                    if (StringUtils.isNotBlank(killTime)) {
-                      val timeoutInSeconds = timeValue.toDouble
-                      val timeoutInMillis = (timeoutInSeconds * 60 * 1000).toLong
-                      if (elapse > timeoutInMillis) {
-                        // 触发kill任务
-                        HttpsUntils.killJob(job)
-                      }
-                    }
-                  }
-                }
-//                }
+              }
+              if (taskMinID == 0L || taskMinID > job.getId) {
+                taskMinID = job.getId
+                scanRuleList.put("jdbcUnfinishedAlertScan", taskMinID)
               }
             case _ =>
               logger.warn(

linkis-extensions/linkis-et-monitor/src/main/scala/org/apache/linkis/monitor/jobhistory/jobtime/StarrocksTimeKillAlertSender.scala

@@ -0,0 +1,31 @@
+/*
+ * 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.linkis.monitor.jobhistory.jobtime
+
+import org.apache.linkis.common.utils.Logging
+import org.apache.linkis.monitor.core.ob.{Event, Observer}
+
+class StarrocksTimeKillAlertSender extends Observer with Logging {
+
+  /**
+   * Observer Pattern
+   */
+  override def update(e: Event, jobHistroyList: scala.Any): Unit = {
+  }
+
+}

linkis-extensions/linkis-et-monitor/src/main/scala/org/apache/linkis/monitor/jobhistory/jobtime/StarrocksTimeKillHitEvent.scala

@@ -0,0 +1,22 @@
+/*
+ * 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.linkis.monitor.jobhistory.jobtime
+
+import org.apache.linkis.monitor.core.ob.SingleObserverEvent
+
+class StarrocksTimeKillHitEvent extends SingleObserverEvent