add support for detach manage mode of flink EC

linkis-computation-governance/linkis-client/linkis-computation-client/src/main/scala/org/apache/linkis/computation/client/once/LinkisManagerClient.scala

@@ -17,14 +17,17 @@
 
 package org.apache.linkis.computation.client.once
 
+import org.apache.linkis.common.utils.Utils
 import org.apache.linkis.computation.client.once.action.{
+  AskEngineConnAction,
   CreateEngineConnAction,
   EngineConnOperateAction,
   GetEngineConnAction,
   KillEngineConnAction,
   LinkisManagerAction
 }
 import org.apache.linkis.computation.client.once.result.{
+  AskEngineConnResult,
   CreateEngineConnResult,
   EngineConnOperateResult,
   GetEngineConnResult,
@@ -39,6 +42,8 @@ import java.io.Closeable
 
 trait LinkisManagerClient extends Closeable {
 
+  def askEngineConn(askEngineConnAction: AskEngineConnAction): AskEngineConnResult
+
   def createEngineConn(createEngineConnAction: CreateEngineConnAction): CreateEngineConnResult
 
   def getEngineConn(getEngineConnAction: GetEngineConnAction): GetEngineConnResult
@@ -82,7 +87,21 @@ class LinkisManagerClientImpl(ujesClient: UJESClient) extends LinkisManagerClien
 
   override def executeEngineConnOperation(
       engineConnOperateAction: EngineConnOperateAction
-  ): EngineConnOperateResult = execute(engineConnOperateAction)
+  ): EngineConnOperateResult = {
+    Utils.tryCatch {
+      val rs = execute[EngineConnOperateResult](engineConnOperateAction)
+      rs
+    } { case e: Exception =>
+      val rs = new EngineConnOperateResult
+      rs.setIsError(true)
+      rs.setErrorMsg(e.getMessage)
+      rs
+    }
+  }
 
   override def close(): Unit = ujesClient.close()
+
+  override def askEngineConn(askEngineConnAction: AskEngineConnAction): AskEngineConnResult =
+    execute(askEngineConnAction)
+
 }

linkis-computation-governance/linkis-client/linkis-computation-client/src/main/scala/org/apache/linkis/computation/client/once/action/AskEngineConnAction.scala

@@ -0,0 +1,101 @@
+/*
+ * 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.computation.client.once.action
+
+import org.apache.linkis.httpclient.dws.DWSHttpClient
+import org.apache.linkis.httpclient.request.POSTAction
+import org.apache.linkis.ujes.client.exception.UJESJobException
+
+import org.apache.commons.lang3.StringUtils
+
+import java.util
+
+class AskEngineConnAction extends POSTAction with LinkisManagerAction {
+
+  override def getRequestPayload: String =
+    DWSHttpClient.jacksonJson.writeValueAsString(getRequestPayloads)
+
+  override def suffixURLs: Array[String] = Array("linkisManager", "askEngineConn")
+
+}
+
+object AskEngineConnAction {
+
+  def newBuilder(): Builder = new Builder
+
+  class Builder private[AskEngineConnAction] () {
+    private var user: String = _
+    private var properties: util.Map[String, String] = _
+    private var labels: util.Map[String, String] = _
+    private var maxSubmitTime: Long = _
+    private var createService: String = _
+    private var description: String = _
+
+    def setUser(user: String): Builder = {
+      this.user = user
+      this
+    }
+
+    def setProperties(properties: util.Map[String, String]): Builder = {
+      this.properties = properties
+      this
+    }
+
+    def setLabels(labels: java.util.Map[String, String]): Builder = {
+      this.labels = labels
+      this
+    }
+
+    def setMaxSubmitTime(maxSubmitTime: Long): Builder = {
+      this.maxSubmitTime = maxSubmitTime
+      this
+    }
+
+    def setCreateService(createService: String): Builder = {
+      this.createService = createService
+      this
+    }
+
+    def setDescription(description: String): Builder = {
+      this.description = description
+      this
+    }
+
+    def build(): AskEngineConnAction = {
+      val action = new AskEngineConnAction()
+      if (user == null) throw new UJESJobException("user is needed!")
+      if (properties == null) properties = new java.util.HashMap[String, String]
+      if (labels == null) throw new UJESJobException("labels is needed!")
+      action.setUser(user)
+      action.addRequestPayload("properties", properties)
+      action.addRequestPayload("labels", labels)
+      if (StringUtils.isNotBlank(createService)) {
+        action.addRequestPayload("createService", createService)
+      }
+      if (null != maxSubmitTime) {
+        action.addRequestPayload("timeOut", maxSubmitTime)
+      }
+      if (StringUtils.isNotBlank(description)) {
+        action.addRequestPayload("description", description)
+      }
+      action
+    }
+
+  }
+
+}

linkis-computation-governance/linkis-client/linkis-computation-client/src/main/scala/org/apache/linkis/computation/client/once/result/AskEngineConnResult.scala

@@ -0,0 +1,23 @@
+/*
+ * 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.computation.client.once.result
+
+import org.apache.linkis.httpclient.dws.annotation.DWSHttpMessageResult
+
+@DWSHttpMessageResult("/api/rest_j/v\\d+/linkisManager/askEngineConn")
+class AskEngineConnResult extends GetEngineConnResult

linkis-computation-governance/linkis-client/linkis-computation-client/src/main/scala/org/apache/linkis/computation/client/once/result/EngineConnOperateResult.scala

@@ -33,9 +33,11 @@ class EngineConnOperateResult extends LinkisManagerResult {
     this.result = result
   }
 
+  def getErrorMsg(): String = errorMsg
+
   def setErrorMsg(errorMsg: String): Unit = this.errorMsg = errorMsg
 
-  def setError(isError: Boolean): Unit = this.isError = isError
+  def getIsError(): Boolean = isError
 
   def setIsError(isError: Boolean): Unit = this.isError = isError
 

linkis-computation-governance/linkis-client/linkis-computation-client/src/main/scala/org/apache/linkis/computation/client/once/simple/SimpleOnceJob.scala

@@ -95,6 +95,10 @@ trait SimpleOnceJob extends OnceJob {
     case operator => operator
   }
 
+  def getEcServiceInstance: ServiceInstance = serviceInstance
+
+  def getEcTicketId: String = ticketId
+
 }
 
 class SubmittableSimpleOnceJob(

linkis-computation-governance/linkis-client/linkis-computation-client/src/main/scala/org/apache/linkis/computation/client/operator/impl/EngineConnApplicationInfoOperator.scala

@@ -19,6 +19,7 @@ package org.apache.linkis.computation.client.operator.impl
 
 import org.apache.linkis.computation.client.once.result.EngineConnOperateResult
 import org.apache.linkis.computation.client.operator.OnceJobOperator
+import org.apache.linkis.governance.common.constant.ec.ECConstants
 import org.apache.linkis.ujes.client.exception.UJESJobException
 
 class EngineConnApplicationInfoOperator extends OnceJobOperator[ApplicationInfo] {
@@ -28,22 +29,22 @@ class EngineConnApplicationInfoOperator extends OnceJobOperator[ApplicationInfo]
   override protected def resultToObject(result: EngineConnOperateResult): ApplicationInfo = {
     ApplicationInfo(
       result
-        .getAsOption("applicationId")
+        .getAsOption(ECConstants.YARN_APPID_NAME_KEY)
         .getOrElse(
           throw new UJESJobException(
             20300,
             s"Cannot get applicationId from EngineConn $getServiceInstance."
           )
         ),
       result
-        .getAsOption("applicationUrl")
+        .getAsOption(ECConstants.YARN_APP_URL_KEY)
         .getOrElse(
           throw new UJESJobException(
             20300,
             s"Cannot get applicationUrl from EngineConn $getServiceInstance."
           )
         ),
-      result.getAs("queue")
+      result.getAs(ECConstants.QUEUE)
     )
   }
 

linkis-computation-governance/linkis-computation-governance-common/src/main/java/org/apache/linkis/governance/common/enums/OnceJobOperationBoundary.java

@@ -0,0 +1,33 @@
+/*
+ * 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.governance.common.enums;
+
+public enum OnceJobOperationBoundary {
+  ECM("ecm"),
+  EC("ec");
+
+  private String name;
+
+  OnceJobOperationBoundary(String name) {
+    this.name = name;
+  }
+
+  public String getName() {
+    return name;
+  }
+}

linkis-computation-governance/linkis-computation-governance-common/src/main/java/org/apache/linkis/governance/common/exception/engineconn/EngineConnExecutorErrorCode.java

@@ -37,4 +37,6 @@ public class EngineConnExecutorErrorCode {
   public static final int SEND_TO_ENTRANCE_ERROR = 40105;
 
   public static final int INIT_EXECUTOR_FAILED = 40106;
+
+  public static final int INVALID_APPLICATION_ID = 40107;
 }

linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/conf/GovernanceCommonConf.scala

@@ -86,7 +86,12 @@ object GovernanceCommonConf {
     CommonVars(envKey, "").getValue
   }
 
+  // value ECConstants.EC_CLIENT_TYPE_ATTACH
+  val EC_APP_MANAGE_MODE =
+    CommonVars("linkis.ec.app.manage.mode", "attach")
+
   val SCALA_PARSE_APPEND_CODE_ENABLED =
     CommonVars("linkis.scala.parse.append.code.enable", true).getValue
 
+
 }

linkis-computation-governance/linkis-computation-governance-common/src/main/scala/org/apache/linkis/governance/common/constant/ec/ECConstants.scala

@@ -35,4 +35,43 @@ object ECConstants {
 
   val YARN_QUEUE_NAME_CONFIG_KEY = "wds.linkis.rm.yarnqueue"
 
+  val QUEUE = "queue"
+
+  val EC_CLIENT_TYPE_ATTACH = "attach"
+
+  val EC_CLIENT_TYPE_DETACH = "detach"
+
+  val YARN_APPID_NAME_KEY = "applicationId"
+
+  val YARN_APP_URL_KEY = "applicationUrl"
+
+  val YARN_APP_NAME_KEY = "appicationName"
+
+  val YARN_MODE_KEY = "yarnMode"
+
+  val EC_SERVICE_INSTANCE_KEY = "serviceInstance"
+
+  val ECM_SERVICE_INSTANCE_KEY = "ecmServiceInstance"
+
+  val MANAGER_SERVICE_INSTANCE_KEY = "managerServiceInstance"
+
+  val NODE_STATUS_KEY = "nodeStatus"
+
+  val EC_LAST_UNLOCK_TIMESTAMP = "lastUnlockTimestamp"
+
+  val YARN_APP_TYPE_LIST_KEY = "yarnAppTypeList"
+
+  val YARN_APP_STATE_LIST_KEY = "yarnAppStateList"
+
+  val YARN_APP_TYPE_KEY = "yarnAppType"
+
+  val YARN_APP_TYPE_SPARK = "spark"
+
+  val YARN_APP_TYPE_FLINK = "flink"
+
+  val EC_OPERATE_LIST = "list"
+
+  val EC_OPERATE_STATUS = "status"
+
+  val YARN_APP_RESULT_LIST_KEY = "yarnAppResultList"
 }

linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/java/org/apache/linkis/engineconn/computation/concurrent/monitor/TimingMonitorService.java

@@ -55,7 +55,7 @@ public class TimingMonitorService implements InitializingBean, Runnable {
 
   @Override
   public void afterPropertiesSet() throws Exception {
-    if (AccessibleExecutorConfiguration.ENGINECONN_SUPPORT_PARALLELISM()) {
+    if ((Boolean) AccessibleExecutorConfiguration.ENGINECONN_SUPPORT_PARALLELISM().getValue()) {
       Utils.defaultScheduler()
           .scheduleAtFixedRate(
               this, 3 * 60 * 1000, MONITOR_INTERVAL.getValue().toLong(), TimeUnit.MILLISECONDS);

linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/service/DefaultNodeHeartbeatMsgManager.scala

@@ -21,10 +21,13 @@ import org.apache.linkis.common.utils.{Logging, Utils}
 import org.apache.linkis.engineconn.acessible.executor.info.NodeHeartbeatMsgManager
 import org.apache.linkis.engineconn.computation.executor.metrics.ComputationEngineConnMetrics
 import org.apache.linkis.engineconn.core.EngineConnObject
-import org.apache.linkis.engineconn.executor.entity.{Executor, SensibleExecutor}
+import org.apache.linkis.engineconn.executor.entity.{Executor, SensibleExecutor, YarnExecutor}
 import org.apache.linkis.governance.common.constant.ec.ECConstants
+import org.apache.linkis.manager.common.entity.enumeration.NodeStatus
 import org.apache.linkis.server.BDPJettyServerHelper
 
+import org.apache.commons.lang3.StringUtils
+
 import org.springframework.stereotype.Component
 
 import java.util
@@ -72,6 +75,22 @@ class DefaultNodeHeartbeatMsgManager extends NodeHeartbeatMsgManager with Loggin
         engineParams.get(ECConstants.YARN_QUEUE_NAME_CONFIG_KEY).asInstanceOf[Object]
       )
     }
+    executor match {
+      case yarnExecutor: YarnExecutor =>
+        if (StringUtils.isNotBlank(yarnExecutor.getQueue)) {
+          msgMap.put(ECConstants.YARN_QUEUE_NAME_KEY, yarnExecutor.getQueue)
+        }
+        if (StringUtils.isNotBlank(yarnExecutor.getApplicationId)) {
+          msgMap.put(ECConstants.YARN_APPID_NAME_KEY, yarnExecutor.getApplicationId)
+        }
+        if (StringUtils.isNotBlank(yarnExecutor.getApplicationURL)) {
+          msgMap.put(ECConstants.YARN_APP_URL_KEY, yarnExecutor.getApplicationURL)
+        }
+        if (StringUtils.isNotBlank(yarnExecutor.getYarnMode)) {
+          msgMap.put(ECConstants.YARN_MODE_KEY, yarnExecutor.getYarnMode)
+        }
+      case _ =>
+    }
     Utils.tryCatch(BDPJettyServerHelper.gson.toJson(msgMap)) { case e: Exception =>
       val msgs = msgMap.asScala
         .map { case (k, v) => if (null == v) s"${k}->null" else s"${k}->${v.toString}" }

linkis-computation-governance/linkis-engineconn/linkis-engineconn-core/src/main/scala/org/apache/linkis/engineconn/core/executor/ExecutorManager.scala

@@ -137,7 +137,7 @@ class LabelExecutorManagerImpl extends LabelExecutorManager with Logging {
   }
 
   protected def getLabelKey(labels: Array[Label[_]]): String =
-    labels.map(_.getStringValue).mkString("&")
+    labels.filter(null != _).map(_.getStringValue).mkString("&")
 
   protected def createExecutor(
       engineCreationContext: EngineCreationContext,