You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kyuubi.apache.org by ya...@apache.org on 2022/03/31 07:57:29 UTC
[incubator-kyuubi] branch master updated: [KYUUBI #2028] Hive Backend Engine - Events support
This is an automated email from the ASF dual-hosted git repository.
yao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-kyuubi.git
The following commit(s) were added to refs/heads/master by this push:
new 9da22e6 [KYUUBI #2028] Hive Backend Engine - Events support
9da22e6 is described below
commit 9da22e66f3b8dc47de49260a94d4241529186169
Author: Min Zhao <zh...@163.com>
AuthorDate: Thu Mar 31 15:57:18 2022 +0800
[KYUUBI #2028] Hive Backend Engine - Events support
### _Why are the changes needed?_
Add events support for hive engine.
### _How was this patch tested?_
- [ ] 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 #2213 from zhaomin1423/hive_event.
Closes #2028
6e39f032 [Min Zhao] [KYUUBI #2028] Hive Backend Engine - Events support
ae274c93 [Min Zhao] [KYUUBI #2028] Hive Backend Engine - Events support
c71f0559 [Min Zhao] [KYUUBI #2028] Hive Backend Engine - Events support
d58aae88 [Min Zhao] [KYUUBI #2028] Hive Backend Engine - Events support
a1c08f09 [Min Zhao] [KYUUBI #2028] Hive Backend Engine - Events support
aba2f93d [Min Zhao] [KYUUBI #2028] Hive Backend Engine - Events support
0f7e1e06 [Min Zhao] add hive event
0bd9d9ee [Min Zhao] add hive event
Authored-by: Min Zhao <zh...@163.com>
Signed-off-by: Kent Yao <ya...@apache.org>
---
.../apache/kyuubi/engine/hive/HiveSQLEngine.scala | 49 +++++++-
.../engine/hive/events/HiveEngineEvent.scala | 60 +++++++++
.../engine/hive/events/HiveOperationEvent.scala | 78 ++++++++++++
.../kyuubi/engine/hive/events/SessionEvent.scala | 59 +++++++++
.../handler/HiveJsonLoggingEventHandler.scala} | 33 ++---
.../engine/hive/operation/ExecuteStatement.scala | 5 +
.../engine/hive/session/HiveSessionImpl.scala | 15 +++
.../hive/event/HiveEventLoggingServiceSuite.scala | 134 +++++++++++++++++++++
8 files changed, 406 insertions(+), 27 deletions(-)
diff --git a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/HiveSQLEngine.scala b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/HiveSQLEngine.scala
index 906b3c6..62f5dc6 100644
--- a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/HiveSQLEngine.scala
+++ b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/HiveSQLEngine.scala
@@ -17,15 +17,23 @@
package org.apache.kyuubi.engine.hive
+import java.net.InetAddress
+
+import scala.util.control.NonFatal
+
import org.apache.hadoop.hive.conf.HiveConf
import org.apache.hadoop.hive.conf.HiveConf.ConfVars
import org.apache.kyuubi.{Logging, Utils}
import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf.{ENGINE_EVENT_JSON_LOG_PATH, ENGINE_EVENT_LOGGERS}
+import org.apache.kyuubi.engine.hive.events.HiveEngineEvent
+import org.apache.kyuubi.engine.hive.events.handler.HiveJsonLoggingEventHandler
+import org.apache.kyuubi.events.{EventBus, EventLoggerType, KyuubiEvent}
import org.apache.kyuubi.ha.HighAvailabilityConf.HA_ZK_CONN_RETRY_POLICY
import org.apache.kyuubi.ha.client.RetryPolicies
import org.apache.kyuubi.service.{AbstractBackendService, AbstractFrontendService, Serverable}
-import org.apache.kyuubi.util.SignalRegister
+import org.apache.kyuubi.util.{KyuubiHadoopUtils, SignalRegister}
class HiveSQLEngine extends Serverable("HiveSQLEngine") {
override val backendService: AbstractBackendService = new HiveBackendService(this)
@@ -46,8 +54,15 @@ object HiveSQLEngine extends Logging {
var currentEngine: Option[HiveSQLEngine] = None
val hiveConf = new HiveConf()
val kyuubiConf = new KyuubiConf()
+ kyuubiConf.set(ENGINE_EVENT_LOGGERS.key, "JSON")
def startEngine(): HiveSQLEngine = {
+ try {
+ initLoggerEventHandler(kyuubiConf)
+ } catch {
+ case NonFatal(e) =>
+ warn(s"Failed to initialize Logger EventHandler: ${e.getMessage}", e)
+ }
kyuubiConf.setIfMissing(KyuubiConf.FRONTEND_THRIFT_BINARY_BIND_PORT, 0)
kyuubiConf.setIfMissing(HA_ZK_CONN_RETRY_POLICY, RetryPolicies.N_TIME.toString)
@@ -72,20 +87,46 @@ object HiveSQLEngine extends Logging {
val engine = new HiveSQLEngine()
info(s"Starting ${engine.getName}")
engine.initialize(kyuubiConf)
+ EventBus.post(HiveEngineEvent(engine))
engine.start()
+ val event = HiveEngineEvent(engine)
+ info(event)
+ EventBus.post(event)
Utils.addShutdownHook(() => engine.stop())
currentEngine = Some(engine)
engine
}
+ private def initLoggerEventHandler(conf: KyuubiConf): Unit = {
+ val hadoopConf = KyuubiHadoopUtils.newHadoopConf(conf)
+ conf.get(ENGINE_EVENT_LOGGERS).map(EventLoggerType.withName).foreach {
+ case EventLoggerType.JSON =>
+ val hostName = InetAddress.getLocalHost.getCanonicalHostName
+ val handler = HiveJsonLoggingEventHandler(
+ s"Hive-$hostName",
+ ENGINE_EVENT_JSON_LOG_PATH,
+ hadoopConf,
+ conf)
+ EventBus.register[KyuubiEvent](handler)
+ case logger =>
+ throw new IllegalArgumentException(s"Unrecognized event logger: $logger")
+
+ }
+ }
+
def main(args: Array[String]): Unit = {
SignalRegister.registerLogger(logger)
try {
startEngine()
} catch {
- case t: Throwable =>
- error(t)
- currentEngine.foreach(_.stop())
+ case t: Throwable => currentEngine match {
+ case Some(engine) =>
+ engine.stop()
+ val event = HiveEngineEvent(engine).copy(diagnostic = t.getMessage)
+ EventBus.post(event)
+ case _ =>
+ error(s"Failed to start Hive SQL engine: ${t.getMessage}.", t)
+ }
}
}
}
diff --git a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/HiveEngineEvent.scala b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/HiveEngineEvent.scala
new file mode 100644
index 0000000..3941cb6
--- /dev/null
+++ b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/HiveEngineEvent.scala
@@ -0,0 +1,60 @@
+/*
+ * 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.hive.events
+
+import org.apache.kyuubi.Utils
+import org.apache.kyuubi.engine.hive.HiveSQLEngine
+import org.apache.kyuubi.events.KyuubiEvent
+import org.apache.kyuubi.service.ServiceState
+import org.apache.kyuubi.service.ServiceState.ServiceState
+
+case class HiveEngineEvent(
+ connectionUrl: String,
+ startTime: Long,
+ endTime: Long,
+ state: ServiceState,
+ diagnostic: String,
+ settings: Map[String, String]) extends KyuubiEvent {
+
+ override def partitions: Seq[(String, String)] =
+ ("day", Utils.getDateFromTimestamp(startTime)) :: Nil
+
+ override def toString: String = {
+ super.toString
+ }
+}
+
+object HiveEngineEvent {
+
+ def apply(engine: HiveSQLEngine): HiveEngineEvent = {
+ engine.getStartTime
+ val connectionUrl =
+ if (engine.getServiceState.equals(ServiceState.LATENT)) {
+ null
+ } else {
+ engine.frontendServices.head.connectionUrl
+ }
+
+ new HiveEngineEvent(
+ connectionUrl = connectionUrl,
+ startTime = engine.getStartTime,
+ endTime = -1L,
+ state = engine.getServiceState,
+ diagnostic = "",
+ settings = engine.getConf.getAll)
+ }
+}
diff --git a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/HiveOperationEvent.scala b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/HiveOperationEvent.scala
new file mode 100644
index 0000000..4e9fe9b
--- /dev/null
+++ b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/HiveOperationEvent.scala
@@ -0,0 +1,78 @@
+/*
+ * 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.hive.events
+
+import org.apache.kyuubi.Utils
+import org.apache.kyuubi.engine.hive.operation.HiveOperation
+import org.apache.kyuubi.events.KyuubiEvent
+
+/**
+ * A [[HiveOperationEvent]] used to tracker the lifecycle of an operation at Hive SQL Engine side.
+ * <ul>
+ * <li>Operation Basis</li>
+ * <li>Operation Live Status</li>
+ * <li>Parent Session Id</li>
+ * </ul>
+ *
+ * @param statementId the unique identifier of a single operation
+ * @param statement the sql that you execute
+ * @param shouldRunAsync the flag indicating whether the query runs synchronously or not
+ * @param state the current operation state
+ * @param eventTime the time when the event created & logged
+ * @param createTime the time for changing to the current operation state
+ * @param startTime the time the query start to time of this operation
+ * @param completeTime time time the query ends
+ * @param exception: caught exception if have
+ * @param sessionId the identifier of the parent session
+ * @param sessionUser the authenticated client user
+ */
+case class HiveOperationEvent(
+ statementId: String,
+ statement: String,
+ shouldRunAsync: Boolean,
+ state: String,
+ eventTime: Long,
+ createTime: Long,
+ startTime: Long,
+ completeTime: Long,
+ exception: Option[Throwable],
+ sessionId: String,
+ sessionUser: String) extends KyuubiEvent {
+
+ override def partitions: Seq[(String, String)] =
+ ("day", Utils.getDateFromTimestamp(createTime)) :: Nil
+}
+
+object HiveOperationEvent {
+
+ def apply(operation: HiveOperation): HiveOperationEvent = {
+ val session = operation.getSession
+ val status = operation.getStatus
+ new HiveOperationEvent(
+ operation.statementId,
+ operation.redactedStatement,
+ operation.shouldRunAsync,
+ status.state.name(),
+ status.lastModified,
+ status.create,
+ status.start,
+ status.completed,
+ status.exception,
+ session.handle.identifier.toString,
+ session.user)
+ }
+}
diff --git a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/SessionEvent.scala b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/SessionEvent.scala
new file mode 100644
index 0000000..acfb03e
--- /dev/null
+++ b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/SessionEvent.scala
@@ -0,0 +1,59 @@
+/*
+ * 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.hive.events
+
+import org.apache.kyuubi.Utils
+import org.apache.kyuubi.engine.hive.session.HiveSessionImpl
+import org.apache.kyuubi.events.KyuubiEvent
+
+/**
+ * Event Tracking for user sessions
+ * @param sessionId the identifier of a session
+ * @param engineId the engine id
+ * @param startTime Start time
+ * @param endTime End time
+ * @param ip Client IP address
+ * @param serverIp Kyuubi Server IP address
+ * @param totalOperations how many queries and meta calls
+ */
+case class SessionEvent(
+ sessionId: String,
+ engineId: String,
+ username: String,
+ ip: String,
+ serverIp: String,
+ startTime: Long,
+ var endTime: Long = -1L,
+ var totalOperations: Int = 0) extends KyuubiEvent {
+
+ override def partitions: Seq[(String, String)] =
+ ("day", Utils.getDateFromTimestamp(startTime)) :: Nil
+
+}
+
+object SessionEvent {
+
+ def apply(session: HiveSessionImpl): SessionEvent = {
+ new SessionEvent(
+ session.handle.identifier.toString,
+ engineId = "",
+ session.user,
+ session.ipAddress,
+ serverIp = session.serverIpAddress(),
+ session.createTime)
+ }
+}
diff --git a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/operation/ExecuteStatement.scala b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/handler/HiveJsonLoggingEventHandler.scala
similarity index 53%
copy from externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/operation/ExecuteStatement.scala
copy to externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/handler/HiveJsonLoggingEventHandler.scala
index 8b32bf8..fc995b6 100644
--- a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/operation/ExecuteStatement.scala
+++ b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/events/handler/HiveJsonLoggingEventHandler.scala
@@ -14,29 +14,16 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
+package org.apache.kyuubi.engine.hive.events.handler
-package org.apache.kyuubi.engine.hive.operation
+import org.apache.hadoop.conf.Configuration
-import scala.collection.JavaConverters._
+import org.apache.kyuubi.config.{ConfigEntry, KyuubiConf}
+import org.apache.kyuubi.events.handler.JsonLoggingEventHandler
-import org.apache.hive.service.cli.operation.Operation
-
-import org.apache.kyuubi.operation.OperationType
-import org.apache.kyuubi.session.Session
-
-class ExecuteStatement(
- session: Session,
- override val statement: String,
- confOverlay: Map[String, String],
- override val shouldRunAsync: Boolean,
- queryTimeout: Long)
- extends HiveOperation(OperationType.EXECUTE_STATEMENT, session) {
- override val internalHiveOperation: Operation = {
- delegatedOperationManager.newExecuteStatementOperation(
- hive,
- statement,
- confOverlay.asJava,
- shouldRunAsync,
- queryTimeout)
- }
-}
+case class HiveJsonLoggingEventHandler(
+ logName: String,
+ logPath: ConfigEntry[String],
+ hadoopConf: Configuration,
+ kyuubiConf: KyuubiConf)
+ extends JsonLoggingEventHandler(logName, logPath, hadoopConf, kyuubiConf)
diff --git a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/operation/ExecuteStatement.scala b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/operation/ExecuteStatement.scala
index 8b32bf8..b4b8c09 100644
--- a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/operation/ExecuteStatement.scala
+++ b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/operation/ExecuteStatement.scala
@@ -21,6 +21,8 @@ import scala.collection.JavaConverters._
import org.apache.hive.service.cli.operation.Operation
+import org.apache.kyuubi.engine.hive.events.HiveOperationEvent
+import org.apache.kyuubi.events.EventBus
import org.apache.kyuubi.operation.OperationType
import org.apache.kyuubi.session.Session
@@ -31,6 +33,7 @@ class ExecuteStatement(
override val shouldRunAsync: Boolean,
queryTimeout: Long)
extends HiveOperation(OperationType.EXECUTE_STATEMENT, session) {
+
override val internalHiveOperation: Operation = {
delegatedOperationManager.newExecuteStatementOperation(
hive,
@@ -39,4 +42,6 @@ class ExecuteStatement(
shouldRunAsync,
queryTimeout)
}
+
+ EventBus.post(HiveOperationEvent(this))
}
diff --git a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/session/HiveSessionImpl.scala b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/session/HiveSessionImpl.scala
index c797430..798b7b7 100644
--- a/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/session/HiveSessionImpl.scala
+++ b/externals/kyuubi-hive-sql-engine/src/main/scala/org/apache/kyuubi/engine/hive/session/HiveSessionImpl.scala
@@ -25,6 +25,9 @@ import org.apache.hive.service.cli.HiveSQLException
import org.apache.hive.service.cli.session.HiveSession
import org.apache.hive.service.rpc.thrift.TProtocolVersion
+import org.apache.kyuubi.engine.hive.events.SessionEvent
+import org.apache.kyuubi.events.EventBus
+import org.apache.kyuubi.operation.{Operation, OperationHandle}
import org.apache.kyuubi.session.{AbstractSession, SessionHandle, SessionManager}
class HiveSessionImpl(
@@ -39,13 +42,25 @@ class HiveSessionImpl(
val hive: HiveSession)
extends AbstractSession(protocol, user, password, ipAddress, conf, sessionManager) {
+ private val sessionEvent = SessionEvent(this)
+
+ def serverIpAddress(): String = serverIpAddress
+
override def open(): Unit = {
val confClone = new HashMap[String, String]()
confClone.putAll(conf.asJava) // pass conf.asScala not support `put` method
hive.open(confClone)
+ EventBus.post(sessionEvent)
+ }
+
+ override protected def runOperation(operation: Operation): OperationHandle = {
+ sessionEvent.totalOperations += 1
+ super.runOperation(operation)
}
override def close(): Unit = {
+ sessionEvent.endTime = System.currentTimeMillis()
+ EventBus.post(sessionEvent)
super.close()
try {
hive.close()
diff --git a/externals/kyuubi-hive-sql-engine/src/test/scala/org/apache/kyuubi/engine/hive/event/HiveEventLoggingServiceSuite.scala b/externals/kyuubi-hive-sql-engine/src/test/scala/org/apache/kyuubi/engine/hive/event/HiveEventLoggingServiceSuite.scala
new file mode 100644
index 0000000..a1d97f0
--- /dev/null
+++ b/externals/kyuubi-hive-sql-engine/src/test/scala/org/apache/kyuubi/engine/hive/event/HiveEventLoggingServiceSuite.scala
@@ -0,0 +1,134 @@
+/*
+ * 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.hive.event
+
+import java.io.{BufferedReader, InputStreamReader}
+import java.net.InetAddress
+import java.nio.file.Paths
+
+import org.apache.commons.lang3.StringUtils
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.{FileSystem, FSDataInputStream, Path}
+
+import org.apache.kyuubi.Utils
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.config.KyuubiConf.ENGINE_EVENT_JSON_LOG_PATH
+import org.apache.kyuubi.engine.hive.HiveSQLEngine
+import org.apache.kyuubi.engine.hive.events.{HiveEngineEvent, HiveOperationEvent, SessionEvent}
+import org.apache.kyuubi.events.JsonProtocol
+import org.apache.kyuubi.operation.HiveJDBCTestHelper
+
+class HiveEventLoggingServiceSuite extends HiveJDBCTestHelper {
+
+ private val kyuubiConf = new KyuubiConf()
+ private val logRoot = kyuubiConf.get(ENGINE_EVENT_JSON_LOG_PATH)
+ private val currentDate = Utils.getDateFromTimestamp(System.currentTimeMillis())
+ private val hostName = InetAddress.getLocalHost.getCanonicalHostName
+
+ override def beforeAll(): Unit = {
+ HiveSQLEngine.startEngine()
+ super.beforeAll()
+ }
+
+ override def afterAll(): Unit = {
+ super.afterAll()
+ val fileSystem: FileSystem = FileSystem.get(new Configuration())
+ val logPath = new Path(logRoot)
+ if (fileSystem.exists(logPath)) {
+ fileSystem.delete(logPath, true)
+ }
+ fileSystem.close()
+ }
+
+ override protected def jdbcUrl: String = {
+ "jdbc:hive2://" + HiveSQLEngine.currentEngine.get.frontendServices.head.connectionUrl + "/;"
+ }
+
+ test("test engine event logging") {
+ val engineEventPath = Paths.get(
+ logRoot,
+ "hive_engine",
+ s"day=$currentDate",
+ s"Hive-$hostName.json")
+ val fileSystem: FileSystem = FileSystem.get(new Configuration())
+ val fs: FSDataInputStream = fileSystem.open(new Path(engineEventPath.toString))
+ val engineEventReader = new BufferedReader(new InputStreamReader(fs))
+ val readEvent =
+ JsonProtocol.jsonToEvent(engineEventReader.readLine(), classOf[HiveEngineEvent])
+ assert(readEvent.isInstanceOf[HiveEngineEvent])
+ }
+
+ test("test session event logging") {
+ withJdbcStatement() { statement =>
+ val catalogs = statement.getConnection.getMetaData.getCatalogs
+ assert(!catalogs.next())
+ val sessionEventPath = Paths.get(
+ logRoot,
+ "session",
+ s"day=$currentDate",
+ s"Hive-$hostName.json")
+ val fileSystem: FileSystem = FileSystem.get(new Configuration())
+ val fs: FSDataInputStream = fileSystem.open(new Path(sessionEventPath.toString))
+ val engineEventReader = new BufferedReader(new InputStreamReader(fs))
+ val readEvent =
+ JsonProtocol.jsonToEvent(engineEventReader.readLine(), classOf[SessionEvent])
+ assert(readEvent.isInstanceOf[SessionEvent])
+ }
+ }
+
+ test("test operation event logging") {
+ withJdbcStatement("hive_engine_test") { statement =>
+ val createTableStatement = "CREATE TABLE hive_engine_test(id int, value string) stored as orc"
+ statement.execute(createTableStatement)
+ val insertStatement = "INSERT INTO hive_engine_test SELECT 1, '2'"
+ statement.execute(insertStatement)
+ val queryStatement = "SELECT ID, VALUE FROM hive_engine_test"
+ statement.executeQuery(queryStatement)
+
+ val operationEventPath = Paths.get(
+ logRoot,
+ "hive_operation",
+ s"day=$currentDate",
+ s"Hive-$hostName.json")
+
+ val fileSystem: FileSystem = FileSystem.get(new Configuration())
+ val fs: FSDataInputStream = fileSystem.open(new Path(operationEventPath.toString))
+ val engineEventReader = new BufferedReader(new InputStreamReader(fs))
+
+ val createTableEvent =
+ JsonProtocol.jsonToEvent(engineEventReader.readLine(), classOf[HiveOperationEvent])
+ assert(createTableEvent.isInstanceOf[HiveOperationEvent])
+ assert(StringUtils.equals(
+ createTableEvent.asInstanceOf[HiveOperationEvent].statement,
+ createTableStatement))
+
+ val insertEvent =
+ JsonProtocol.jsonToEvent(engineEventReader.readLine(), classOf[HiveOperationEvent])
+ assert(insertEvent.isInstanceOf[HiveOperationEvent])
+ assert(StringUtils.equals(
+ insertEvent.asInstanceOf[HiveOperationEvent].statement,
+ insertStatement))
+
+ val queryEvent =
+ JsonProtocol.jsonToEvent(engineEventReader.readLine(), classOf[HiveOperationEvent])
+ assert(queryEvent.isInstanceOf[HiveOperationEvent])
+ assert(StringUtils.equals(
+ queryEvent.asInstanceOf[HiveOperationEvent].statement,
+ queryStatement))
+ }
+ }
+}