Support engine operator

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

@@ -19,8 +19,8 @@ package org.apache.linkis.computation.client.once
 
 import java.io.Closeable
 
-import org.apache.linkis.computation.client.once.action.{CreateEngineConnAction, GetEngineConnAction, KillEngineConnAction, LinkisManagerAction}
-import org.apache.linkis.computation.client.once.result.{CreateEngineConnResult, GetEngineConnResult, KillEngineConnResult, LinkisManagerResult}
+import org.apache.linkis.computation.client.once.action.{CreateEngineConnAction, EngineOperateAction, GetEngineConnAction, KillEngineConnAction, LinkisManagerAction}
+import org.apache.linkis.computation.client.once.result.{CreateEngineConnResult, EngineOperateResult, GetEngineConnResult, KillEngineConnResult, LinkisManagerResult}
 import org.apache.linkis.httpclient.dws.DWSHttpClient
 import org.apache.linkis.httpclient.request.Action
 import org.apache.linkis.ujes.client.{UJESClient, UJESClientImpl}
@@ -34,6 +34,8 @@ trait LinkisManagerClient extends Closeable {
 
   def killEngineConn(killEngineConnAction: KillEngineConnAction): KillEngineConnResult
 
+  def executeEngineOperation(engineOperateAction: EngineOperateAction): EngineOperateResult
+
 }
 object LinkisManagerClient {
 
@@ -59,5 +61,7 @@ class LinkisManagerClientImpl(ujesClient: UJESClient) extends LinkisManagerClien
 
   override def killEngineConn(killEngineConnAction: KillEngineConnAction): KillEngineConnResult = execute(killEngineConnAction)
 
+  override def executeEngineOperation(engineOperateAction: EngineOperateAction): EngineOperateResult = execute(engineOperateAction)
+
   override def close(): Unit = ujesClient.close()
 }

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

@@ -0,0 +1,67 @@
+/*
+ * 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 java.util
+
+class EngineOperateAction extends GetEngineConnAction {
+
+  override def suffixURLs: Array[String] = Array("linkisManager", "executeEngineOperation")
+
+}
+
+object EngineOperateAction {
+
+  val OPERATOR_NAME_KEY = "__operator_name__"
+
+  def newBuilder(): Builder = new Builder
+
+  class Builder extends ServiceInstanceBuilder[EngineOperateAction] {
+
+    private var operatorName = ""
+
+    private var properties: util.Map[String, Any] = new util.HashMap[String, Any]
+
+    def operatorName(operatorName: String): this.type  = {
+      this.operatorName = operatorName
+      this
+    }
+
+    def setProperties(properties: util.Map[String, Any]): this.type = {
+      this.properties = properties
+      this
+    }
+
+    def addProperty(key: String, value: Any): this.type = {
+      if (this.properties == null) {
+        this.properties = new util.HashMap[String, Any]
+      }
+      this.properties.put(key, value)
+      this
+    }
+
+    override protected def createGetEngineConnAction(): EngineOperateAction = {
+      val action = new EngineOperateAction
+      addProperty(OPERATOR_NAME_KEY, operatorName)
+      action.addRequestPayload("properties", properties)
+      action
+    }
+  }
+
+}
+

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

@@ -0,0 +1,44 @@
+/*
+ * 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 java.util
+
+import org.apache.linkis.httpclient.dws.annotation.DWSHttpMessageResult
+
+@DWSHttpMessageResult("/api/rest_j/v\\d+/linkisManager/executeEngineOperation")
+class EngineOperateResult extends LinkisManagerResult {
+
+  private var result: util.Map[String, Any] = _
+
+  def setResult(result: util.Map[String, Any]): Unit = {
+    this.result = result
+  }
+
+  def getResult: util.Map[String, Any] = result
+
+  def getAs[T](key: String): Option[T] = {
+    if (result != null && result.get(key) != null) {
+      Some(result.get(key).asInstanceOf[T])
+    } else {
+      None
+    }
+  }
+
+
+}

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

@@ -91,7 +91,7 @@ class SubmittableSimpleOnceJob(protected override val linkisManagerClient: Linki
     info(s"EngineConn created with status $lastEngineConnState, the nodeInfo is $lastNodeInfo.")
     addOperatorAction {
       case onceJobOperator: OnceJobOperator[_] =>
-        onceJobOperator.setServiceInstance(serviceInstance).setLinkisManagerClient(linkisManagerClient)
+        onceJobOperator.setUser(user).setServiceInstance(serviceInstance).setLinkisManagerClient(linkisManagerClient)
       case operator => operator
     }
     if(!isCompleted(lastEngineConnState) && !isRunning) {

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

@@ -23,12 +23,19 @@ import org.apache.linkis.computation.client.once.LinkisManagerClient
 
 trait OnceJobOperator[T] extends Operator[T] {
 
+  private var user: String = _
   private var serviceInstance: ServiceInstance = _
   private var linkisManagerClient: LinkisManagerClient = _
 
+  protected def getUser: String = user
   protected def getServiceInstance: ServiceInstance = serviceInstance
   protected def getLinkisManagerClient: LinkisManagerClient = linkisManagerClient
 
+  def setUser(user: String): this.type = {
+    this.user = user
+    this
+  }
+
   def setServiceInstance(serviceInstance: ServiceInstance): this.type = {
     this.serviceInstance = serviceInstance
     this

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

@@ -17,6 +17,7 @@
  */
 package org.apache.linkis.computation.client.operator.impl
 
+import org.apache.linkis.computation.client.once.action.EngineOperateAction
 import org.apache.linkis.computation.client.operator.OnceJobOperator
 
 
@@ -25,8 +26,20 @@ class ApplicationInfoOperator extends OnceJobOperator[ApplicationInfo] {
   override def getName: String = ApplicationInfoOperator.OPERATOR_NAME
 
   override def apply(): ApplicationInfo = {
-    //TODO
-    ApplicationInfo("", "", "")
+
+    val engineOperateAction = EngineOperateAction.newBuilder()
+      .operatorName(getName)
+      .setUser(getUser)
+      .setApplicationName(getServiceInstance.getApplicationName)
+      .setInstance(getServiceInstance.getInstance)
+      .build()
+
+    val result = getLinkisManagerClient.executeEngineOperation(engineOperateAction)
+    ApplicationInfo(
+      result.getAs("applicationId").getOrElse(""),
+      result.getAs("applicationUrl").getOrElse(""),
+      result.getAs("queue").getOrElse("")
+    )
   }
 
 }

linkis-computation-governance/linkis-engineconn/linkis-engineconn-executor/accessible-executor/pom.xml

@@ -33,12 +33,6 @@
             <groupId>org.apache.linkis</groupId>
             <artifactId>linkis-executor-core</artifactId>
             <version>${linkis.version}</version>
-            <exclusions>
-                <exclusion>
-                    <artifactId>reflections</artifactId>
-                    <groupId>org.reflections</groupId>
-                </exclusion>
-            </exclusions>
         </dependency>
 
         <dependency>

linkis-computation-governance/linkis-engineconn/linkis-engineconn-executor/accessible-executor/src/main/scala/org/apache/linkis/engineconn/acessible/executor/operator/Operator.scala

@@ -0,0 +1,30 @@
+/*
+ * 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.engineconn.acessible.executor.operator
+
+import org.apache.linkis.manager.common.protocol.engine.EngineOperateResponse
+
+trait Operator {
+
+  def getName: String
+
+  def init(properties: Map[String, Any])
+
+  def apply(): EngineOperateResponse
+
+}

linkis-computation-governance/linkis-engineconn/linkis-engineconn-executor/accessible-executor/src/main/scala/org/apache/linkis/engineconn/acessible/executor/operator/OperatorFactory.scala

@@ -0,0 +1,56 @@
+/*
+ * 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.engineconn.acessible.executor.operator
+
+import org.apache.linkis.common.exception.WarnException
+import org.apache.linkis.common.utils.ClassUtils
+import org.apache.linkis.manager.common.protocol.engine.EngineOperateRequest
+
+trait OperatorFactory {
+
+  def createOperatorRequest(request: EngineOperateRequest): Operator
+
+}
+
+object OperatorFactory {
+
+  private val operatorFactory = new OperatorFactoryImpl
+
+  def apply(): OperatorFactory = operatorFactory
+
+}
+
+import scala.collection.convert.WrapAsScala._
+class OperatorFactoryImpl extends OperatorFactory {
+
+  private val operatorClasses: Map[String, Class[_ <: Operator]] = ClassUtils.reflections.getSubTypesOf(classOf[Operator])
+    .filterNot(ClassUtils.isInterfaceOrAbstract).map { clazz =>
+      clazz.newInstance().getName -> clazz
+    }.toMap
+
+  override def createOperatorRequest(request: EngineOperateRequest): Operator = {
+    request.properties.getOrElse(EngineOperateRequest.OPERATOR_NAME_KEY, null) match {
+      case operatorName: String if operatorClasses.contains(operatorName) =>
+        val operator = operatorClasses.get(operatorName).get.newInstance()
+        operator.init(request.properties)
+        operator
+      case _ => throw new WarnException(-1, s"Cannot find operator.")
+    }
+  }
+
+}

linkis-computation-governance/linkis-engineconn/linkis-engineconn-executor/accessible-executor/src/main/scala/org/apache/linkis/engineconn/acessible/executor/service/DefaultOperateService.scala

@@ -0,0 +1,32 @@
+/*
+ * 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.engineconn.acessible.executor.service
+import org.apache.linkis.engineconn.acessible.executor.operator.OperatorFactory
+import org.apache.linkis.manager.common.protocol.engine.{EngineOperateRequest, EngineOperateResponse}
+import org.apache.linkis.message.annotation.Receiver
+import org.springframework.stereotype.Service
+
+@Service
+class DefaultOperateService extends OperateService {
+
+  @Receiver
+  override def executeOperation(engineOperateRequest: EngineOperateRequest): EngineOperateResponse = {
+    val operator = OperatorFactory().createOperatorRequest(engineOperateRequest)
+    operator()
+  }
+}

linkis-computation-governance/linkis-engineconn/linkis-engineconn-executor/accessible-executor/src/main/scala/org/apache/linkis/engineconn/acessible/executor/service/OperateService.scala

@@ -0,0 +1,26 @@
+/*
+ * 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.engineconn.acessible.executor.service
+
+import org.apache.linkis.manager.common.protocol.engine.{EngineOperateRequest, EngineOperateResponse}
+
+trait OperateService {
+
+  def executeOperation(engineOperateRequest: EngineOperateRequest): EngineOperateResponse
+
+}