Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[KYUUBI #5797][FOLLOWUP] Desc engine command support show engine registered attributes #5948

Closed
wants to merge 6 commits into from
Closed

[KYUUBI #5797][FOLLOWUP] Desc engine command support show engine registered attributes #5948

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
9dfb2f5
[KYUUBI #5797][FOLLOWUP] desc engine command support show engine regi…
Jan 5, 2024
9fcc2c6
delete ENGINE_NAMESPACE column
Jan 5, 2024
cca773a
fix code
Jan 16, 2024
d58e8ec
delete empty lines
Jan 16, 2024
085ffc5
fix code
Jan 18, 2024
da89099
fix code
Jan 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions kyuubi-server/src/main/scala/org/apache/kyuubi/client/KyuubiSyncThriftClient.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,7 @@ import org.apache.kyuubi.util.{ThreadUtils, ThriftUtils}
import org.apache.kyuubi.util.ThreadUtils.scheduleTolerableRunnableWithFixedDelay

class KyuubiSyncThriftClient private (
hostPort: (String, Int),
beryllw marked this conversation as resolved.
Show resolved Hide resolved
protocol: TProtocol,
engineAliveProbeProtocol: Option[TProtocol],
engineAliveProbeInterval: Long,
Expand All @@ -52,6 +53,7 @@ class KyuubiSyncThriftClient private (
@volatile private var _engineId: Option[String] = _
@volatile private var _engineUrl: Option[String] = _
@volatile private var _engineName: Option[String] = _
private[kyuubi] def engineHostPort: (String, Int) = hostPort
beryllw marked this conversation as resolved.
Show resolved Hide resolved

private[kyuubi] def engineConnectionClosed: Boolean = !protocol.getTransport.isOpen

Expand Down Expand Up @@ -483,6 +485,7 @@ private[kyuubi] object KyuubiSyncThriftClient extends Logging {
None
}
new KyuubiSyncThriftClient(
(host, port),
tProtocol,
aliveProbeProtocol,
aliveProbeInterval,
Expand Down
11 changes: 10 additions & 1 deletion kyuubi-server/src/main/scala/org/apache/kyuubi/engine/EngineRef.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,7 @@ import org.apache.kyuubi.engine.jdbc.JdbcProcessBuilder
import org.apache.kyuubi.engine.spark.SparkProcessBuilder
import org.apache.kyuubi.engine.trino.TrinoProcessBuilder
import org.apache.kyuubi.ha.HighAvailabilityConf.{HA_ENGINE_REF_ID, HA_NAMESPACE}
import org.apache.kyuubi.ha.client.{DiscoveryClient, DiscoveryClientProvider, DiscoveryPaths}
import org.apache.kyuubi.ha.client.{DiscoveryClient, DiscoveryClientProvider, DiscoveryPaths, ServiceNodeInfo}
import org.apache.kyuubi.metrics.MetricsConstants.{ENGINE_FAIL, ENGINE_TIMEOUT, ENGINE_TOTAL}
import org.apache.kyuubi.metrics.MetricsSystem
import org.apache.kyuubi.operation.log.OperationLog
Expand Down Expand Up @@ -337,6 +337,15 @@ private[kyuubi] class EngineRef(
}
}

def getServiceNodes(
beryllw marked this conversation as resolved.
Show resolved Hide resolved
discoveryClient: DiscoveryClient,
hostPort: (String, Int)): Seq[ServiceNodeInfo] = {
tryWithLock(discoveryClient) {
beryllw marked this conversation as resolved.
Show resolved Hide resolved
val serviceNodes = discoveryClient.getServiceNodesInfo(engineSpace)
serviceNodes.filter { sn => (sn.host, sn.port) == hostPort }
}
}

beryllw marked this conversation as resolved.
Show resolved Hide resolved
def close(): Unit = {
if (shareLevel == CONNECTION && builder != null) {
try {
Expand Down
7 changes: 7 additions & 0 deletions kyuubi-server/src/main/scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@ import org.apache.kyuubi.config.KyuubiReservedKeys.{KYUUBI_ENGINE_CREDENTIALS_KE
import org.apache.kyuubi.engine.{EngineRef, KyuubiApplicationManager}
import org.apache.kyuubi.events.{EventBus, KyuubiSessionEvent}
import org.apache.kyuubi.ha.client.DiscoveryClientProvider._
import org.apache.kyuubi.ha.client.ServiceNodeInfo
import org.apache.kyuubi.operation.{Operation, OperationHandle}
import org.apache.kyuubi.operation.log.OperationLog
import org.apache.kyuubi.service.authentication.InternalSecurityAccessor
Expand Down Expand Up @@ -119,6 +120,12 @@ class KyuubiSessionImpl(
engineLastAlive = System.currentTimeMillis()
}

def listZkEngineNodes: Seq[ServiceNodeInfo] = {
beryllw marked this conversation as resolved.
Show resolved Hide resolved
withDiscoveryClient(sessionConf) { discoveryClient =>
engine.getServiceNodes(discoveryClient, _client.engineHostPort)
}
}

private[kyuubi] def openEngineSession(extraEngineLog: Option[OperationLog] = None): Unit =
handleSessionException {
withDiscoveryClient(sessionConf) { discoveryClient =>
Expand Down
24 changes: 16 additions & 8 deletions kyuubi-server/src/main/scala/org/apache/kyuubi/sql/plan/command/DescribeEngine.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,13 +35,18 @@ import org.apache.kyuubi.sql.schema.{Column, Row, Schema}
case class DescribeEngine() extends RunnableCommand {

override def run(kyuubiSession: KyuubiSession): Unit = {
val rows = Seq(kyuubiSession).map { session =>
lazy val client = session.asInstanceOf[KyuubiSessionImpl].client
val values = new ListBuffer[String]()
values += client.engineId.getOrElse("")
values += client.engineName.getOrElse("")
values += client.engineUrl.getOrElse("")
Row(values.toList)
val rows = Seq(kyuubiSession.asInstanceOf[KyuubiSessionImpl]).flatMap { session =>
lazy val client = session.client
session.listZkEngineNodes.map { nodeInfo =>
val values = new ListBuffer[String]()
values += client.engineId.getOrElse("")
values += client.engineName.getOrElse("")
values += client.engineUrl.getOrElse("")
values += s"${nodeInfo.host}:${nodeInfo.port}"
values += nodeInfo.version.getOrElse("")
values += nodeInfo.attributes.filter(_._1.contains("memory")).mkString(";")
beryllw marked this conversation as resolved.
Show resolved Hide resolved
Row(values.toList)
}
}
iter = new IterableFetchIterator(rows)
}
Expand All @@ -59,6 +64,9 @@ object DescribeEngine {
Seq(
Column("ENGINE_ID", TTypeId.STRING_TYPE, Some("Kyuubi engine identify")),
Column("ENGINE_NAME", TTypeId.STRING_TYPE, Some("Kyuubi engine name")),
Column("ENGINE_URL", TTypeId.STRING_TYPE, Some("Kyuubi engine url")))
Column("ENGINE_URL", TTypeId.STRING_TYPE, Some("Kyuubi engine url")),
Column("ENGINE_INSTANCE", TTypeId.STRING_TYPE, Some("Kyuubi engine instance host and port")),
Column("ENGINE_VERSION", TTypeId.STRING_TYPE, Some("Kyuubi engine version")),
Column("ENGINE_ATTRIBUTES", TTypeId.STRING_TYPE, Some("Kyuubi engine attributes")))
}
}
5 changes: 4 additions & 1 deletion kyuubi-server/src/test/scala/org/apache/kyuubi/operation/parser/DescribeEngineSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,13 @@ class DescribeEngineSuite extends ExecutedCommandExecSuite {
val resultSet = statement.executeQuery(s"KYUUBI $desc ENGINE")
assert(resultSet.next())

assert(resultSet.getMetaData.getColumnCount == 3)
assert(resultSet.getMetaData.getColumnCount == 6)
assert(resultSet.getMetaData.getColumnName(1) == "ENGINE_ID")
assert(resultSet.getMetaData.getColumnName(2) == "ENGINE_NAME")
assert(resultSet.getMetaData.getColumnName(3) == "ENGINE_URL")
assert(resultSet.getMetaData.getColumnName(4) == "ENGINE_INSTANCE")
assert(resultSet.getMetaData.getColumnName(5) == "ENGINE_VERSION")
assert(resultSet.getMetaData.getColumnName(6) == "ENGINE_ATTRIBUTES")
}
}
}
Expand Down
Loading