[1.9.0]Added StarRocks task timeout alarm and timeout kill function

linkis-extensions/linkis-et-monitor/src/main/java/org/apache/linkis/monitor/config/ListenerConfig.java

@@ -17,7 +17,7 @@
 
 package org.apache.linkis.monitor.config;
 
-import org.apache.linkis.monitor.until.HttpsUntils;
+import org.apache.linkis.monitor.entity.ClientSingleton;
 import org.apache.linkis.monitor.until.ThreadUtils;
 import org.apache.linkis.monitor.utils.log.LogUtils;
 
@@ -38,7 +38,7 @@ public class ListenerConfig {
   private void shutdownEntrance(ContextClosedEvent event) {
     try {
       ThreadUtils.executors.shutdown();
-      HttpsUntils.client.close();
+      ClientSingleton.getInstance().close();
     } catch (IOException e) {
       logger.error("ListenerConfig error msg  {}", e.getMessage());
     }

linkis-extensions/linkis-et-monitor/src/main/java/org/apache/linkis/monitor/entity/ClientSingleton.java

@@ -0,0 +1,107 @@
+/*
+ * 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.entity;
+
+import org.apache.linkis.bml.conf.BmlConfiguration;
+import org.apache.linkis.common.conf.Configuration;
+import org.apache.linkis.httpclient.dws.authentication.TokenAuthenticationStrategy;
+import org.apache.linkis.httpclient.dws.config.DWSClientConfig;
+import org.apache.linkis.httpclient.dws.config.DWSClientConfigBuilder;
+import org.apache.linkis.monitor.client.MonitorHTTPClient;
+import org.apache.linkis.monitor.client.MonitorHTTPClientClientImpl;
+
+import org.apache.commons.collections.MapUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+public class ClientSingleton {
+  private static MonitorHTTPClient instance;
+  private static DWSClientConfig dwsClientConfig;
+
+  private ClientSingleton() {}
+
+  public static synchronized MonitorHTTPClient getInstance() {
+    if (instance == null) {
+      if (dwsClientConfig == null) {
+        dwsClientConfig = createClientConfig(null, null);
+      }
+      instance = new MonitorHTTPClientClientImpl(dwsClientConfig);
+    }
+    return instance;
+  }
+
+  public static DWSClientConfig createClientConfig(String url, Map<String, Object> properties) {
+    String realUrl = "";
+    if (StringUtils.isBlank(url)) {
+      realUrl = Configuration.getGateWayURL();
+    } else {
+      realUrl = url;
+    }
+    Map<String, Object> parms = new HashMap<>();
+    if (MapUtils.isNotEmpty(properties)) {
+      parms = properties;
+    }
+    int maxConnection =
+        (int)
+            parms.getOrDefault(
+                BmlConfiguration.CONNECTION_MAX_SIZE_SHORT_NAME(),
+                BmlConfiguration.CONNECTION_MAX_SIZE().getValue());
+    int connectTimeout =
+        (int)
+            parms.getOrDefault(
+                BmlConfiguration.CONNECTION_TIMEOUT_SHORT_NAME(),
+                BmlConfiguration.CONNECTION_TIMEOUT().getValue());
+    int readTimeout =
+        (int)
+            parms.getOrDefault(
+                BmlConfiguration.CONNECTION_READ_TIMEOUT_SHORT_NAME(),
+                BmlConfiguration.CONNECTION_READ_TIMEOUT().getValue());
+    String tokenKey =
+        (String)
+            parms.getOrDefault(
+                BmlConfiguration.AUTH_TOKEN_KEY_SHORT_NAME(),
+                BmlConfiguration.AUTH_TOKEN_KEY().getValue());
+    String tokenValue =
+        (String)
+            parms.getOrDefault(
+                BmlConfiguration.AUTH_TOKEN_VALUE_SHORT_NAME(),
+                BmlConfiguration.AUTH_TOKEN_VALUE().getValue());
+
+    DWSClientConfig clientConfig =
+        ((DWSClientConfigBuilder)
+                (DWSClientConfigBuilder.newBuilder()
+                    .addServerUrl(realUrl)
+                    .connectionTimeout(connectTimeout)
+                    .discoveryEnabled(false)
+                    .discoveryFrequency(1, TimeUnit.MINUTES)
+                    .loadbalancerEnabled(false)
+                    .maxConnectionSize(maxConnection)
+                    .retryEnabled(false)
+                    .readTimeout(readTimeout)
+                    .setAuthenticationStrategy(new TokenAuthenticationStrategy())
+                    .setAuthTokenKey(tokenKey)
+                    .setAuthTokenValue(tokenValue)))
+            .setDWSVersion("v1")
+            .build();
+
+    return clientConfig;
+  }
+}

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

@@ -29,6 +29,8 @@
 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.labels.JobHistoryLabelsAlertSender;
 import org.apache.linkis.monitor.jobhistory.labels.JobHistoryLabelsRule;
 import org.apache.linkis.monitor.jobhistory.runtime.CommonJobRunTimeRule;
@@ -204,15 +206,18 @@ public void jobHistoryUnfinishedScan() {
       logger.info("[INFO] Loaded 0 alerts jobtime alert-rule from alert properties file.");
     } else {
       logger.info(
-          "[INFO] Loaded {} alerts jobtime alert-rules from alert properties file.",
-          jobTimeAlerts.size());
+              "[INFO] Loaded {} alerts jobtime alert-rules from alert properties file.",
+              jobTimeAlerts.size());
       shouldStart = true;
       JobMonitorUtils.addIntervalToImsAlerts(jobTimeAlerts, realIntervals);
       JobTimeExceedRule jobTimeExceedRule =
-          new JobTimeExceedRule(
-              jobTimeAlerts.keySet(), new JobTimeExceedAlertSender(jobTimeAlerts));
-      scanner.addScanRule(jobTimeExceedRule);
+              new JobTimeExceedRule(
+                      jobTimeAlerts.keySet(), new JobTimeExceedAlertSender(jobTimeAlerts));
+            scanner.addScanRule(jobTimeExceedRule);
     }
+    StarrocksTimeExceedRule starrocksTimeExceedRule =
+        new StarrocksTimeExceedRule(new StarrocksTimeExceedAlterSender());
+    scanner.addScanRule(starrocksTimeExceedRule);
     JobMonitorUtils.run(scanner, fetchers, shouldStart);
   }
 }

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

@@ -61,7 +61,7 @@ public void ecmResourceTask() {
     // 获取emNode资源信息
     List<Map<String, Object>> emNodeVoList = new ArrayList<>();
     try {
-      Map<String, Object> resultmap = HttpsUntils.sendHttp(null, null);
+      Map<String, Object> resultmap = HttpsUntils.getEmsResourceList(null, null);
       // got interface data
       Map<String, List<Map<String, Object>>> data = MapUtils.getMap(resultmap, "data");
       emNodeVoList = data.getOrDefault("EMs", new ArrayList<>());

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

@@ -24,7 +24,7 @@
 import org.apache.linkis.manager.label.constant.LabelKeyConstant;
 import org.apache.linkis.monitor.config.MonitorConfig;
 import org.apache.linkis.monitor.constants.Constants;
-import org.apache.linkis.monitor.until.HttpsUntils;
+import org.apache.linkis.monitor.entity.ClientSingleton;
 import org.apache.linkis.monitor.utils.alert.AlertDesc;
 import org.apache.linkis.monitor.utils.alert.ims.MonitorAlertUtils;
 import org.apache.linkis.monitor.utils.alert.ims.PooledImsAlertUtils;
@@ -57,7 +57,8 @@ public class UserModeMonitor {
 
   private static final Logger logger = LoggerFactory.getLogger(UserModeMonitor.class);
 
-  private static final DWSClientConfig clientConfig = HttpsUntils.dwsClientConfig;
+  private static final DWSClientConfig clientConfig =
+      ClientSingleton.createClientConfig(null, null);
 
   private static final UJESClient client = new UJESClientImpl(clientConfig);
 
@@ -138,7 +139,7 @@ private static JobExecuteResult toSubmit(LinkedTreeMap<String, String> engine) {
   public void dbJob() {
     Map<String, Object> properties = new HashMap<>();
     properties.put("readTimeout", MonitorConfig.USER_MODE_INTERFACE_TIMEOUT.getValue());
-    DWSClientConfig clientConfig = HttpsUntils.createClientConfig(null, properties);
+    DWSClientConfig clientConfig = ClientSingleton.createClientConfig(null, properties);
     UJESClientImpl ujesClient = new UJESClientImpl(clientConfig);
     GetTableStatisticInfoAction builder =
         GetTableStatisticInfoAction.builder()

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

@@ -17,24 +17,22 @@
 
 package org.apache.linkis.monitor.until;
 
-import org.apache.linkis.bml.conf.BmlConfiguration;
-import org.apache.linkis.common.conf.Configuration;
 import org.apache.linkis.common.utils.Utils;
-import org.apache.linkis.httpclient.dws.authentication.TokenAuthenticationStrategy;
-import org.apache.linkis.httpclient.dws.config.DWSClientConfig;
-import org.apache.linkis.httpclient.dws.config.DWSClientConfigBuilder;
+import org.apache.linkis.datasource.client.response.GetInfoPublishedByDataSourceNameResult;
 import org.apache.linkis.monitor.client.MonitorHTTPClient;
-import org.apache.linkis.monitor.client.MonitorHTTPClientClientImpl;
 import org.apache.linkis.monitor.config.MonitorConfig;
+import org.apache.linkis.monitor.entity.ClientSingleton;
 import org.apache.linkis.monitor.entity.IndexEntity;
-import org.apache.linkis.monitor.request.EmsListAction;
-import org.apache.linkis.monitor.request.EntranceTaskAction;
+import org.apache.linkis.monitor.jobhistory.entity.JobHistory;
+import org.apache.linkis.monitor.request.*;
 import org.apache.linkis.monitor.response.EntranceTaskResult;
+import org.apache.linkis.monitor.response.KeyvalueResult;
+import org.apache.linkis.monitor.response.KillJobResultAction;
+import org.apache.linkis.protocol.utils.ZuulEntranceUtils;
 import org.apache.linkis.server.BDPJettyServerHelper;
 import org.apache.linkis.ujes.client.response.EmsListResult;
 
 import org.apache.commons.collections.MapUtils;
-import org.apache.commons.lang3.StringUtils;
 import org.apache.http.client.config.RequestConfig;
 import org.apache.http.client.methods.CloseableHttpResponse;
 import org.apache.http.client.methods.HttpPost;
@@ -47,100 +45,27 @@
 import org.springframework.util.Assert;
 
 import java.io.IOException;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
+import java.util.*;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class HttpsUntils {
   private static final Logger logger = LoggerFactory.getLogger(HttpsUntils.class);
 
-  public static DWSClientConfig dwsClientConfig = createClientConfig(null, null);
-  //        IOUtils.closeQuietly(client);
-  public static MonitorHTTPClient client = new MonitorHTTPClientClientImpl(dwsClientConfig);
   public static final String localHost = Utils.getLocalHostname();
 
-  public static Map<String, Object> sendHttp(String url, Map<String, Object> properties)
+  public static Map<String, Object> getEmsResourceList(String url, Map<String, Object> properties)
       throws IOException {
-    if (null == dwsClientConfig) {
-      dwsClientConfig = createClientConfig(url, properties);
-    }
-    if (null == client) {
-      client = new MonitorHTTPClientClientImpl(dwsClientConfig);
-    }
+    MonitorHTTPClient client = ClientSingleton.getInstance();
     EmsListAction build = EmsListAction.newBuilder().setUser("hadoop").build();
     EmsListResult result = client.list(build);
     return result.getResultMap();
   }
 
-  public static DWSClientConfig createClientConfig(String url, Map<String, Object> properties) {
-    String realUrl = "";
-    if (StringUtils.isBlank(url)) {
-      realUrl = Configuration.getGateWayURL();
-    } else {
-      realUrl = url;
-    }
-    Map<String, Object> parms = new HashMap<>();
-    if (MapUtils.isNotEmpty(properties)) {
-      parms = properties;
-    }
-    int maxConnection =
-        (int)
-            parms.getOrDefault(
-                BmlConfiguration.CONNECTION_MAX_SIZE_SHORT_NAME(),
-                BmlConfiguration.CONNECTION_MAX_SIZE().getValue());
-    int connectTimeout =
-        (int)
-            parms.getOrDefault(
-                BmlConfiguration.CONNECTION_TIMEOUT_SHORT_NAME(),
-                BmlConfiguration.CONNECTION_TIMEOUT().getValue());
-    int readTimeout =
-        (int)
-            parms.getOrDefault(
-                BmlConfiguration.CONNECTION_READ_TIMEOUT_SHORT_NAME(),
-                BmlConfiguration.CONNECTION_READ_TIMEOUT().getValue());
-    String tokenKey =
-        (String)
-            parms.getOrDefault(
-                BmlConfiguration.AUTH_TOKEN_KEY_SHORT_NAME(),
-                BmlConfiguration.AUTH_TOKEN_KEY().getValue());
-    String tokenValue =
-        (String)
-            parms.getOrDefault(
-                BmlConfiguration.AUTH_TOKEN_VALUE_SHORT_NAME(),
-                BmlConfiguration.AUTH_TOKEN_VALUE().getValue());
-
-    DWSClientConfig clientConfig =
-        ((DWSClientConfigBuilder)
-                (DWSClientConfigBuilder.newBuilder()
-                    .addServerUrl(realUrl)
-                    .connectionTimeout(connectTimeout)
-                    .discoveryEnabled(false)
-                    .discoveryFrequency(1, TimeUnit.MINUTES)
-                    .loadbalancerEnabled(false)
-                    .maxConnectionSize(maxConnection)
-                    .retryEnabled(false)
-                    .readTimeout(readTimeout)
-                    .setAuthenticationStrategy(new TokenAuthenticationStrategy())
-                    .setAuthTokenKey(tokenKey)
-                    .setAuthTokenValue(tokenValue)))
-            .setDWSVersion("v1")
-            .build();
-
-    return clientConfig;
-  }
-
   public static Map<String, Object> getEntranceTask(String url, String user, String Instance)
       throws IOException {
-    if (null == dwsClientConfig) {
-      dwsClientConfig = createClientConfig(null, null);
-    }
-    if (null == client) {
-      client = new MonitorHTTPClientClientImpl(dwsClientConfig);
-    }
+    MonitorHTTPClient client = ClientSingleton.getInstance();
     EntranceTaskAction build =
         EntranceTaskAction.newBuilder().setUser(user).setInstance(Instance).build();
     EntranceTaskResult result = client.entranList(build);
@@ -170,4 +95,50 @@ public static void sendIndex(List<IndexEntity> list) throws IOException {
     logger.info("send index response :{}", map);
     Assert.isTrue(!"0".equals(map.get("resultCode")), map.get("resultMsg"));
   }
+
+  public static String getJDBCConf(String user, String conf) {
+    MonitorHTTPClient client = ClientSingleton.getInstance();
+    KeyvalueAction build =
+        KeyvalueAction.newBuilder()
+            .setVersion("4")
+            .setEngineType("jdbc")
+            .setCreator("IDE")
+            .setConfigKey(conf)
+            .setUser(user)
+            .build();
+    KeyvalueResult result = client.getConfKeyValue(build);
+    Map data = MapUtils.getMap(result.getResultMap(), "data", new HashMap<>());
+    Map configValues = MapUtils.getMap(data, "configValues", new HashMap<>());
+    return MapUtils.getString(configValues, "configValue", "");
+  }
+
+  public static Map getDatasourceConf(String user, String datasourceName) {
+    MonitorHTTPClient client = ClientSingleton.getInstance();
+    DataSourceParamsAction dataSourceParamsAction =
+        DataSourceParamsAction.builder()
+            .setSystem("5435")
+            .setDataSourceName(datasourceName)
+            .setUser(user)
+            .build();
+    GetInfoPublishedByDataSourceNameResult result = client.getInfoByDataSourceInfo(dataSourceParamsAction);
+    Map data = MapUtils.getMap(result.getResultMap(), "data", new HashMap<>());
+    Map datasourceInfoMap = MapUtils.getMap(data, "info", new HashMap<>());
+    return datasourceInfoMap;
+  }
+
+  public static void killJob(JobHistory jobHistory) {
+    MonitorHTTPClient client = ClientSingleton.getInstance();
+    String[] split = jobHistory.getInstances().split(";");
+    String execID =
+        ZuulEntranceUtils.generateExecID(jobHistory.getJobReqId(), "linkis-cg-entrance", split);
+    KillJobAction killJobAction =
+        KillJobAction.builder()
+            .setIdList(Collections.singletonList(execID))
+            .setTaskIDList(Collections.singletonList(jobHistory.getId()))
+            .setExecID(execID)
+            .setUser(jobHistory.getSubmitUser())
+            .build();
+    KillJobResultAction killJobResultAction = client.killJob(killJobAction);
+    Map data = MapUtils.getMap(killJobResultAction.getResultMap(), "data", new HashMap<>());
+  }
 }

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

@@ -47,7 +47,7 @@
 
     <sql id="jobhistory_query">
         job.`id`,job.`job_req_id`,job.`submit_user`,job.`execute_user`,job.`labels`,job.`params`,job.`status`,job.`error_code`,job.`created_time`,
-        job.`updated_time`,job.`instances`,job.`observe_info`,org.`org_id`,org.`org_name`
+        job.`updated_time`,job.`instances`,job.`engine_type`,job.`observe_info`,org.`org_id`,org.`org_name`
     </sql>
 
     <select id="selectJobHistory" useCache="false" resultMap="jobHistoryMap"