Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[KYUUBI #2445] Implement ApplicationManager and Yarn/ JPS-local Appli…
…cation Operation ### _Why are the changes needed?_ Add KyuubiApplicationManager in SessionManager for application management, currently support kill and get application information. The underlying cluster manager operation added in this PR are - local jps with SIG TERM KILL - YARN-client ### _How was this patch tested?_ - [x] Add some test cases that check the changes thoroughly including negative and positive cases if possible - [ ] Add screenshots for manual tests if appropriate - [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request Closes #2447 from yaooqinn/2445. Closes #2445 7781039 [Kent Yao] address comment aed3f25 [Kent Yao] address comment 0fad941 [Kent Yao] address comment 67ec250 [Kent Yao] address comment 800bedb [Kent Yao] [KYUUBI #2445] Implement ApplicationManager and Yarn/ JPS-local Application Operation 1a4084a [Kent Yao] [KYUUBI #2445] Implement ApplicationManager and Yarn/ JPS-local Application Operation be58583 [Kent Yao] [KYUUBI #2445] Implement ApplicationManager and Yarn/ JPS-local Application Operation e75e20e [Kent Yao] [KYUUBI #2445] Implement ApplicationManager and Yarn/ JPS-local Application Operation baac7f0 [Kent Yao] [KYUUBI #2445] Implement ApplicationManager and Yarn/ JPS-local Application Operation e3f5c29 [Kent Yao] [KYUUBI #2445] Implement ApplicationManager and Yarn/ JPS-local Application Operation c81e563 [Kent Yao] [KYUUBI #2445] Implement ApplicationManager and Yarn/ JPS-local Application Operation Authored-by: Kent Yao <yao@apache.org> Signed-off-by: Kent Yao <yao@apache.org>
- Loading branch information
Showing
27 changed files
with
704 additions
and
440 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
19 changes: 19 additions & 0 deletions
19
...server/src/main/resources/META-INF/services/org.apache.kyuubi.engine.ApplicationOperation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,19 @@ | ||
| # | ||
| # 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. | ||
| # | ||
|
|
||
| org.apache.kyuubi.engine.YarnApplicationOperation | ||
| org.apache.kyuubi.engine.JpsApplicationOperation |
75 changes: 75 additions & 0 deletions
75
kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ApplicationOperation.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,75 @@ | ||
| /* | ||
| * 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.kyuubi.engine | ||
|
|
||
| import org.apache.kyuubi.config.KyuubiConf | ||
|
|
||
| trait ApplicationOperation { | ||
|
|
||
| /** | ||
| * Step for initializing the instance. | ||
| */ | ||
| def initialize(conf: KyuubiConf): Unit | ||
|
|
||
| /** | ||
| * Step to clean up the instance | ||
| */ | ||
| def stop(): Unit | ||
|
|
||
| /** | ||
| * Called before other method to do a quick skip | ||
| * | ||
| * @param clusterManager the underlying cluster manager or just local instance | ||
| */ | ||
| def isSupported(clusterManager: Option[String]): Boolean | ||
|
|
||
| /** | ||
| * Kill the app/engine by the unique application tag | ||
| * | ||
| * @param tag the unique application tag for engine instance. | ||
| * For example, | ||
| * if the Hadoop Yarn is used, for spark applications, | ||
| * the tag will be preset via spark.yarn.tags | ||
| * @return a message contains response describing how the kill process. | ||
| * | ||
| * @note For implementations, please suppress exceptions and always return KillResponse | ||
| */ | ||
| def killApplicationByTag(tag: String): KillResponse | ||
|
|
||
| /** | ||
| * Get the engine/application status by the unique application tag | ||
| * | ||
| * @param tag the unique application tag for engine instance. | ||
| * @return a map contains the application status | ||
| */ | ||
| def getApplicationInfoByTag(tag: String): Map[String, String] | ||
| } | ||
|
|
||
| object ApplicationOperation { | ||
|
|
||
| /** | ||
| * identifier determined by cluster manager for the engine | ||
| */ | ||
| val APP_ID_KEY = "id" | ||
| val APP_NAME_KEY = "name" | ||
| val APP_STATE_KEY = "state" | ||
| val APP_URL_KEY = "url" | ||
| val APP_ERROR_KEY = "error" | ||
|
|
||
| val NOT_FOUND = "APPLICATION_NOT_FOUND" | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
92 changes: 92 additions & 0 deletions
92
kyuubi-server/src/main/scala/org/apache/kyuubi/engine/JpsApplicationOperation.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,92 @@ | ||
| /* | ||
| * 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.kyuubi.engine | ||
|
|
||
| import java.nio.file.Paths | ||
|
|
||
| import scala.sys.process._ | ||
|
|
||
| import org.apache.kyuubi.config.KyuubiConf | ||
| import org.apache.kyuubi.engine.ApplicationOperation.NOT_FOUND | ||
|
|
||
| class JpsApplicationOperation extends ApplicationOperation { | ||
|
|
||
| private var runner: String = _ | ||
|
|
||
| override def initialize(conf: KyuubiConf): Unit = { | ||
| val jps = sys.env.get("JAVA_HOME").orElse(sys.props.get("java.home")) | ||
| .map(Paths.get(_, "bin", "jps").toString) | ||
| .getOrElse("jps") | ||
| runner = | ||
| try { | ||
| jps.!! | ||
| } catch { | ||
| case _: Throwable => null | ||
| } | ||
| } | ||
|
|
||
| override def isSupported(clusterManager: Option[String]): Boolean = { | ||
| runner != null && (clusterManager.isEmpty || clusterManager.get == "local") | ||
| } | ||
|
|
||
| private def getEngine(tag: String): Option[String] = { | ||
| if (runner == null) { | ||
| None | ||
| } else { | ||
| val pb = "jps -ml" #| s"grep $tag" | ||
| try { | ||
| pb.lineStream_!.headOption | ||
| } catch { | ||
| case _: Throwable => None | ||
| } | ||
| } | ||
| } | ||
|
|
||
| override def killApplicationByTag(tag: String): KillResponse = { | ||
| val commandOption = getEngine(tag) | ||
| if (commandOption.nonEmpty) { | ||
| val idAndCmd = commandOption.get | ||
| val (id, _) = idAndCmd.splitAt(idAndCmd.indexOf(" ")) | ||
| try { | ||
| s"kill -15 $id".lineStream | ||
| (true, s"Succeeded to terminate: $idAndCmd") | ||
| } catch { | ||
| case e: Exception => | ||
| (false, s"Failed to terminate: $idAndCmd, due to ${e.getMessage}") | ||
| } | ||
| } else { | ||
| (false, NOT_FOUND) | ||
| } | ||
| } | ||
|
|
||
| override def getApplicationInfoByTag(tag: String): Map[String, String] = { | ||
| val commandOption = getEngine(tag) | ||
| if (commandOption.nonEmpty) { | ||
| val idAndCmd = commandOption.get | ||
| val (id, cmd) = idAndCmd.splitAt(idAndCmd.indexOf(" ")) | ||
| Map( | ||
| "id" -> id, | ||
| "name" -> cmd, | ||
| "state" -> "RUNNING") | ||
| } else { | ||
| Map("state" -> "FINISHED") | ||
| } | ||
| } | ||
|
|
||
| override def stop(): Unit = {} | ||
| } |
Oops, something went wrong.